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

equal deleted inserted replaced

-:4db4bec0e5bb
+:8a6065d830b9
 import java.io.EOFException;
 import java.lang.System.Logger.Level;
 import java.nio.ByteBuffer;
 import java.util.concurrent.CompletableFuture;
-import java.util.concurrent.CompletionStage;
 import java.util.concurrent.Executor;
-import java.util.function.BiConsumer;
+import java.util.concurrent.atomic.AtomicLong;
 import java.util.function.Consumer;
 import java.util.function.Function;
 import java.net.http.HttpHeaders;
 import java.net.http.HttpResponse;
 import jdk.internal.net.http.ResponseContent.BodyParser;
 // Revisit: can we get rid of this?
 static enum State {INITIAL, READING_HEADERS, READING_BODY, DONE}
 private volatile State readProgress = State.INITIAL;
 static final boolean DEBUG = Utils.DEBUG; // Revisit: temporary dev flag.
-final System.Logger  debug = Utils.getDebugLogger(this.getClass()::getSimpleName, DEBUG);
+final System.Logger  debug = Utils.getDebugLogger(this::dbgString, DEBUG);
+final static AtomicLong responseCount = new AtomicLong();
+final long id = responseCount.incrementAndGet();
 Http1Response(HttpConnection conn,
 Http1Exchange<T> exchange,
 Http1AsyncReceiver asyncReceiver) {
 this.readProgress = State.INITIAL;
 this.exchange = exchange;
 this.connection = conn;
 this.asyncReceiver = asyncReceiver;
 headersReader = new HeadersReader(this::advance);
 bodyReader = new BodyReader(this::advance);
+}
+String dbgTag;
+private String dbgString() {
+String dbg = dbgTag;
+if (dbg == null) {
+String cdbg = connection.dbgTag;
+if (cdbg != null) {
+dbgTag = dbg = "Http1Response(id=" + id + ", " + cdbg + ")";
+} else {
+dbg = "Http1Response(id=" + id + ")";
+}
+}
+return dbg;
+}
+// The ClientRefCountTracker is used to track the state
+// of a pending operation. Altough there usually is a single
+// point where the operation starts, it may terminate at
+// different places.
+private final class ClientRefCountTracker {
+final HttpClientImpl client = connection.client();
+// state & 0x01 != 0 => acquire called
+// state & 0x02 != 0 => tryRelease called
+byte state;
+public synchronized void acquire() {
+if (state == 0) {
+// increment the reference count on the HttpClientImpl
+// to prevent the SelectorManager thread from exiting
+// until our operation is complete.
+debug.log(Level.DEBUG, "incrementing ref count for %s", client);
+client.reference();
+state = 0x01;
+} else {
+assert (state & 0x01) == 0 : "reference count already incremented";
+}
+}
+public synchronized void tryRelease() {
+if (state == 0x01) {
+// decrement the reference count on the HttpClientImpl
+// to allow the SelectorManager thread to exit if no
+// other operation is pending and the facade is no
+// longer referenced.
+debug.log(Level.DEBUG, "decrementing ref count for %s", client);
+client.unreference();
+state |= 0x02;
+}
+}
 }
 public CompletableFuture<Response> readHeadersAsync(Executor executor) {
 debug.log(Level.DEBUG, () -> "Reading Headers: (remaining: "
 + asyncReceiver.remaining() +") "  + readProgress);
 } else {
 return readBody(HttpResponse.BodySubscriber.discard(), true, executor);
 }
 }
 public <U> CompletableFuture<U> readBody(HttpResponse.BodySubscriber<U> p,
 boolean return2Cache,
 Executor executor) {
 this.return2Cache = return2Cache;
 final HttpResponse.BodySubscriber<U> pusher = p;
 // expect-continue reads headers and body twice.
 // if we reach here, we must reset the headersReader state.
 asyncReceiver.unsubscribe(headersReader);
 headersReader.reset();
+ClientRefCountTracker refCountTracker = new ClientRefCountTracker();
 executor.execute(() -> {
 try {
-HttpClientImpl client = connection.client();
 content = new ResponseContent(
 connection, clen, headers, pusher,
 this::onFinished
 );
 if (cf.isCompletedExceptionally()) {
 return;
 }
 // increment the reference count on the HttpClientImpl
 // to prevent the SelectorManager thread from exiting until
 // the body is fully read.
-client.reference();
+refCountTracker.acquire();
 bodyReader.start(content.getBodyParser(
 (t) -> {
 try {
 if (t != null) {
 pusher.onError(t);
 connection.close();
 if (!cf.isDone())
 cf.completeExceptionally(t);
 }
 } finally {
-// decrement the reference count on the HttpClientImpl
-// to allow the SelectorManager thread to exit if no
-// other operation is pending and the facade is no
-// longer referenced.
-client.unreference();
 bodyReader.onComplete(t);
 }
 }));
 CompletableFuture<State> bodyReaderCF = bodyReader.completion();
 asyncReceiver.subscribe(bodyReader);
 // Make sure to keep a reference to asyncReceiver from
 // within this
 CompletableFuture<?> trailingOp = bodyReaderCF.whenComplete((s,t) ->  {
 t = Utils.getCompletionCause(t);
 try {
-if (t != null) {
+if (t == null) {
 debug.log(Level.DEBUG, () ->
 "Finished reading body: " + s);
 assert s == State.READING_BODY;
 }
 if (t != null && !cf.isDone()) {
 cf.completeExceptionally(t);
 }
 } catch (Throwable x) {
 // not supposed to happen
 asyncReceiver.onReadError(x);
+} finally {
+// we're done: release the ref count for
+// the current operation.
+refCountTracker.tryRelease();
 }
 });
 connection.addTrailingOperation(trailingOp);
 } catch (Throwable t) {
 debug.log(Level.DEBUG, () -> "Failed reading body: " + t);
 } finally {
 asyncReceiver.onReadError(t);
 }
 }
 });
-p.getBody().whenComplete((U u, Throwable t) -> {
+try {
-if (t == null)
+p.getBody().whenComplete((U u, Throwable t) -> {
-cf.complete(u);
+if (t == null)
-else
+cf.complete(u);
-cf.completeExceptionally(t);
+else
+cf.completeExceptionally(t);
+});
+} catch (Throwable t) {
+cf.completeExceptionally(t);
+asyncReceiver.setRetryOnError(false);
+asyncReceiver.onReadError(t);
+}
+return cf.whenComplete((s,t) -> {
+if (t != null) {
+// If an exception occurred, release the
+// ref count for the current operation, as
+// it may never be triggered otherwise
+// (BodySubscriber ofInputStream)
+// If there was no exception then the
+// ref count will be/have been released when
+// the last byte of the response is/was received
+refCountTracker.tryRelease();
+}
 });
-return cf;
 }
 private void onFinished() {
 asyncReceiver.clear();

branch	http-client-branch
changeset 56165	8a6065d830b9
parent 56126	86e628130926
child 56167	96fa4f49a9ff