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

equal deleted inserted replaced

-:818a23db260c
+:4254bed3c09d
 package jdk.internal.net.http;
 import javax.net.ssl.SSLContext;
 import javax.net.ssl.SSLParameters;
 import java.io.IOException;
+import java.io.UncheckedIOException;
 import java.lang.ref.Reference;
 import java.lang.ref.WeakReference;
 import java.net.Authenticator;
+import java.net.ConnectException;
 import java.net.CookieHandler;
 import java.net.ProxySelector;
+import java.net.http.HttpTimeoutException;
 import java.nio.ByteBuffer;
 import java.nio.channels.CancelledKeyException;
 import java.nio.channels.ClosedChannelException;
 import java.nio.channels.SelectableChannel;
 import java.nio.channels.SelectionKey;
 import java.util.Objects;
 import java.util.Optional;
 import java.util.Set;
 import java.util.TreeSet;
 import java.util.concurrent.CompletableFuture;
-import java.util.concurrent.ConcurrentLinkedQueue;
+import java.util.concurrent.CompletionException;
 import java.util.concurrent.ExecutionException;
 import java.util.concurrent.Executor;
 import java.util.concurrent.Executors;
 import java.util.concurrent.ThreadFactory;
 import java.util.concurrent.atomic.AtomicInteger;
 import java.util.concurrent.atomic.AtomicLong;
+import java.util.function.BooleanSupplier;
 import java.util.stream.Stream;
 import java.net.http.HttpClient;
 import java.net.http.HttpRequest;
 import java.net.http.HttpResponse;
 import java.net.http.HttpResponse.BodyHandler;
 t.setDaemon(true);
 return t;
 }
 }
+/**
+* A DelegatingExecutor is an executor that delegates tasks to
+* a wrapped executor when it detects that the current thread
+* is the SelectorManager thread. If the current thread is not
+* the selector manager thread the given task is executed inline.
+*/
+final static class DelegatingExecutor implements Executor {
+private final BooleanSupplier isInSelectorThread;
+private final Executor delegate;
+DelegatingExecutor(BooleanSupplier isInSelectorThread, Executor delegate) {
+this.isInSelectorThread = isInSelectorThread;
+this.delegate = delegate;
+}
+Executor delegate() {
+return delegate;
+}
+@Override
+public void execute(Runnable command) {
+if (isInSelectorThread.getAsBoolean()) {
+delegate.execute(command);
+} else {
+command.run();
+}
+}
+}
 private final CookieHandler cookieHandler;
 private final Redirect followRedirects;
 private final Optional<ProxySelector> userProxySelector;
 private final ProxySelector proxySelector;
 private final Authenticator authenticator;
 private final Version version;
 private final ConnectionPool connections;
-private final Executor executor;
+private final DelegatingExecutor delegatingExecutor;
 private final boolean isDefaultExecutor;
 // Security parameters
 private final SSLContext sslContext;
 private final SSLParameters sslParams;
 private final SelectorManager selmgr;
 Executor ex = builder.executor;
 if (ex == null) {
 ex = Executors.newCachedThreadPool(new DefaultThreadFactory(id));
 isDefaultExecutor = true;
 } else {
-ex = builder.executor;
 isDefaultExecutor = false;
 }
+delegatingExecutor = new DelegatingExecutor(this::isSelectorThread, ex);
 facadeRef = new WeakReference<>(facadeFactory.createFacade(this));
 client2 = new Http2ClientImpl(this);
-executor = ex;
 cookieHandler = builder.cookieHandler;
 followRedirects = builder.followRedirects == null ?
 Redirect.NEVER : builder.followRedirects;
 this.userProxySelector = Optional.ofNullable(builder.proxy);
 this.proxySelector = userProxySelector
 @Override
 public <T> HttpResponse<T>
 send(HttpRequest req, BodyHandler<T> responseHandler)
 throws IOException, InterruptedException
 {
+CompletableFuture<HttpResponse<T>> cf = null;
 try {
-return sendAsync(req, responseHandler, null).get();
+cf = sendAsync(req, responseHandler, null, null);
+return cf.get();
+} catch (InterruptedException ie) {
+if (cf != null )
+cf.cancel(true);
+throw ie;
 } catch (ExecutionException e) {
-Throwable t = e.getCause();
+final Throwable throwable = e.getCause();
-if (t instanceof Error)
+final String msg = throwable.getMessage();
-throw (Error)t;
-if (t instanceof RuntimeException)
+if (throwable instanceof IllegalArgumentException) {
-throw (RuntimeException)t;
+throw new IllegalArgumentException(msg, throwable);
-else if (t instanceof IOException)
+} else if (throwable instanceof SecurityException) {
-throw Utils.getIOException(t);
+throw new SecurityException(msg, throwable);
-else
+} else if (throwable instanceof HttpTimeoutException) {
-throw new InternalError("Unexpected exception", t);
+throw new HttpTimeoutException(msg);
-}
+} else if (throwable instanceof ConnectException) {
-}
+ConnectException ce = new ConnectException(msg);
+ce.initCause(throwable);
+throw ce;
+} else if (throwable instanceof IOException) {
+throw new IOException(msg, throwable);
+} else {
+throw new IOException(msg, throwable);
+}
+}
+}
+private static final Executor ASYNC_POOL = new CompletableFuture<Void>().defaultExecutor();
 @Override
 public <T> CompletableFuture<HttpResponse<T>>
 sendAsync(HttpRequest userRequest, BodyHandler<T> responseHandler)
 {
 return sendAsync(userRequest, responseHandler, null);
 }
 @Override
 public <T> CompletableFuture<HttpResponse<T>>
 sendAsync(HttpRequest userRequest,
 BodyHandler<T> responseHandler,
-PushPromiseHandler<T> pushPromiseHandler)
+PushPromiseHandler<T> pushPromiseHandler) {
-{
+return sendAsync(userRequest, responseHandler, pushPromiseHandler, delegatingExecutor.delegate);
+}
+private <T> CompletableFuture<HttpResponse<T>>
+sendAsync(HttpRequest userRequest,
+BodyHandler<T> responseHandler,
+PushPromiseHandler<T> pushPromiseHandler,
+Executor exchangeExecutor)    {
 Objects.requireNonNull(userRequest);
 Objects.requireNonNull(responseHandler);
 AccessControlContext acc = null;
 if (System.getSecurityManager() != null)
 reference();
 try {
 if (debugelapsed.on())
 debugelapsed.log("ClientImpl (async) send %s", userRequest);
-Executor executor = acc == null
+// When using sendAsync(...) we explicitly pass the
-? this.executor
+// executor's delegate as exchange executor to force
-: new PrivilegedExecutor(this.executor, acc);
+// asynchronous scheduling of the exchange.
+// When using send(...) we don't specify any executor
+// and default to using the client's delegating executor
+// which only spawns asynchronous tasks if it detects
+// that the current thread is the selector manager
+// thread. This will cause everything to execute inline
+// until we need to schedule some event with the selector.
+Executor executor = exchangeExecutor == null
+? this.delegatingExecutor : exchangeExecutor;
 MultiExchange<T> mex = new MultiExchange<>(userRequest,
 requestImpl,
 this,
 responseHandler,
 pushPromiseHandler,
 acc);
 CompletableFuture<HttpResponse<T>> res =
-mex.responseAsync().whenComplete((b,t) -> unreference());
+mex.responseAsync(executor).whenComplete((b,t) -> unreference());
 if (DEBUGELAPSED) {
 res = res.whenComplete(
 (b,t) -> debugCompleted("ClientImpl (async)", start, userRequest));
 }
-// makes sure that any dependent actions happen in the executor
+// makes sure that any dependent actions happen in the CF default
-res = res.whenCompleteAsync((r, t) -> { /* do nothing */}, executor);
+// executor. This is only needed for sendAsync(...), when
+// exchangeExecutor is non-null.
+if (exchangeExecutor != null) {
+res = res.whenCompleteAsync((r, t) -> { /* do nothing */}, ASYNC_POOL);
+}
 return res;
 } catch(Throwable t) {
 unreference();
 debugCompleted("ClientImpl (async)", start, userRequest);
 throw t;
 void wakeupSelector() {
 selector.wakeup();
 }
 synchronized void shutdown() {
+Log.logTrace("{0}: shutting down", getName());
 if (debug.on()) debug.log("SelectorManager shutting down");
 closed = true;
 try {
 selector.close();
 } catch (IOException ignored) {
 public void run() {
 List<Pair<AsyncEvent,IOException>> errorList = new ArrayList<>();
 List<AsyncEvent> readyList = new ArrayList<>();
 List<Runnable> resetList = new ArrayList<>();
 try {
+if (Log.channel()) Log.logChannel(getName() + ": starting");
 while (!Thread.currentThread().isInterrupted()) {
 synchronized (this) {
 assert errorList.isEmpty();
 assert readyList.isEmpty();
 assert resetList.isEmpty();
 sa.register(event);
 if (!chan.isOpen()) {
 throw new IOException("Channel closed");
 }
 } catch (IOException e) {
-Log.logTrace("HttpClientImpl: " + e);
+Log.logTrace("{0}: {1}", getName(), e);
 if (debug.on())
 debug.log("Got " + e.getClass().getName()
 + " while handling registration events");
 chan.close();
 // let the event abort deal with it
 errorList.clear();
 // Check whether client is still alive, and if not,
 // gracefully stop this thread
 if (!owner.isReferenced()) {
-Log.logTrace("HttpClient no longer referenced. Exiting...");
+Log.logTrace("{0}: {1}",
+getName(),
+"HttpClient no longer referenced. Exiting...");
 return;
 }
 // Timeouts will have milliseconds granularity. It is important
 // to handle them in a timely fashion.
 int n = selector.select(millis == 0 ? NODEADLINE : millis);
 if (n == 0) {
 // Check whether client is still alive, and if not,
 // gracefully stop this thread
 if (!owner.isReferenced()) {
-Log.logTrace("HttpClient no longer referenced. Exiting...");
+Log.logTrace("{0}: {1}",
+getName(),
+"HttpClient no longer referenced. Exiting...");
 return;
 }
 owner.purgeTimeoutsAndReturnNextDeadline();
 continue;
 }
 resetList.forEach(r -> r.run());
 resetList.clear();
 }
 } catch (Throwable e) {
-//e.printStackTrace();
 if (!closed) {
 // This terminates thread. So, better just print stack trace
 String err = Utils.stackTrace(e);
-Log.logError("HttpClientImpl: fatal error: " + err);
+Log.logError("{0}: {1}: {2}", getName(),
+"HttpClientImpl shutting down due to fatal error", err);
 }
 if (debug.on()) debug.log("shutting down", e);
 if (Utils.ASSERTIONSENABLED && !debug.on()) {
 e.printStackTrace(System.err); // always print the stack
 }
 } finally {
+if (Log.channel()) Log.logChannel(getName() + ": stopping");
 shutdown();
 }
 }
 //        void debugPrint(Selector selector) {
 @Override
 public Optional<Authenticator> authenticator() {
 return Optional.ofNullable(authenticator);
 }
-/*package-private*/ final Executor theExecutor() {
+/*package-private*/ final DelegatingExecutor theExecutor() {
-return executor;
+return delegatingExecutor;
 }
 @Override
 public final Optional<Executor> executor() {
-return isDefaultExecutor ? Optional.empty() : Optional.of(executor);
+return isDefaultExecutor
+? Optional.empty()
+: Optional.of(delegatingExecutor.delegate());
 }
 ConnectionPool connectionPool() {
 return connections;
 }

changeset 50681	4254bed3c09d
parent 49944	4690a2871b44
child 50985	cd41f34e548c
child 56795	03ece2518428
child 56799	c274589ad63b