/* * Copyright (c) 2015, 2025, 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. Oracle designates this * particular file as subject to the "Classpath" exception as provided * by Oracle in the LICENSE file that accompanied this code. * * 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. */ package jdk.internal.net.http; import javax.net.ssl.SSLContext; import javax.net.ssl.SSLException; import javax.net.ssl.SSLHandshakeException; 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.InetAddress; import java.net.ProtocolException; import java.net.ProxySelector; import java.net.http.HttpConnectTimeoutException; import java.net.http.HttpTimeoutException; import java.nio.ByteBuffer; import java.nio.channels.CancelledKeyException; import java.nio.channels.ClosedChannelException; import java.nio.channels.ClosedSelectorException; import java.nio.channels.SelectableChannel; import java.nio.channels.SelectionKey; import java.nio.channels.Selector; import java.security.NoSuchAlgorithmException; import java.time.Duration; import java.time.temporal.ChronoUnit; import java.util.ArrayList; import java.util.HashSet; import java.util.Iterator; import java.util.List; 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.ConcurrentSkipListSet; import java.util.concurrent.ExecutionException; import java.util.concurrent.Executor; import java.util.concurrent.ExecutorService; import java.util.concurrent.Executors; import java.util.concurrent.RejectedExecutionException; import java.util.concurrent.ThreadFactory; import java.util.concurrent.atomic.AtomicBoolean; import java.util.concurrent.atomic.AtomicInteger; import java.util.concurrent.atomic.AtomicLong; import java.util.concurrent.atomic.AtomicReference; import java.util.concurrent.locks.ReentrantLock; import java.util.function.BiConsumer; 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; import java.net.http.HttpResponse.PushPromiseHandler; import java.net.http.WebSocket; import jdk.internal.net.http.common.BufferSupplier; import jdk.internal.net.http.common.Deadline; import jdk.internal.net.http.common.HttpBodySubscriberWrapper; import jdk.internal.net.http.common.Log; import jdk.internal.net.http.common.Logger; import jdk.internal.net.http.common.MinimalFuture; import jdk.internal.net.http.common.Pair; import jdk.internal.net.http.common.TimeSource; import jdk.internal.net.http.common.Utils; import jdk.internal.net.http.common.OperationTrackers.Trackable; import jdk.internal.net.http.common.OperationTrackers.Tracker; import jdk.internal.net.http.websocket.BuilderImpl; /** * Client implementation. Contains all configuration information and also * the selector manager thread which allows async events to be registered * and delivered when they occur. See AsyncEvent. */ final class HttpClientImpl extends HttpClient implements Trackable { static final boolean DEBUGELAPSED = Utils.TESTING || Utils.DEBUG; // dev flag static final boolean DEBUGTIMEOUT = false; // dev flag final Logger debug = Utils.getDebugLogger(this::dbgString, Utils.DEBUG); final Logger debugelapsed = Utils.getDebugLogger(this::dbgString, DEBUGELAPSED); final Logger debugtimeout = Utils.getDebugLogger(this::dbgString, DEBUGTIMEOUT); static final AtomicLong CLIENT_IDS = new AtomicLong(); private final AtomicLong CONNECTION_IDS = new AtomicLong(); static final int DEFAULT_KEEP_ALIVE_TIMEOUT = 30; static final long KEEP_ALIVE_TIMEOUT = getTimeoutProp("jdk.httpclient.keepalive.timeout", DEFAULT_KEEP_ALIVE_TIMEOUT); // Defaults to value used for HTTP/1 Keep-Alive Timeout. Can be overridden by jdk.httpclient.keepalive.timeout.h2 property. static final long IDLE_CONNECTION_TIMEOUT = getTimeoutProp("jdk.httpclient.keepalive.timeout.h2", KEEP_ALIVE_TIMEOUT); // Define the default factory as a static inner class // that embeds all the necessary logic to avoid // the risk of using a lambda that might keep a reference on the // HttpClient instance from which it was created (helps with // heapdump analysis). private static final class DefaultThreadFactory implements ThreadFactory { private final String namePrefix; private final AtomicInteger nextId = new AtomicInteger(); DefaultThreadFactory(long clientID) { namePrefix = "HttpClient-" + clientID + "-Worker-"; } @Override public Thread newThread(Runnable r) { String name = namePrefix + nextId.getAndIncrement(); Thread t = new Thread(null, r, name, 0, false); 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. */ static final class DelegatingExecutor implements Executor { private final BooleanSupplier isInSelectorThread; private final Executor delegate; private final BiConsumer errorHandler; DelegatingExecutor(BooleanSupplier isInSelectorThread, Executor delegate, BiConsumer errorHandler) { this.isInSelectorThread = isInSelectorThread; this.delegate = delegate; this.errorHandler = errorHandler; } Executor delegate() { return delegate; } @Override public void execute(Runnable command) { if (isInSelectorThread.getAsBoolean()) { ensureExecutedAsync(command); } else { command.run(); } } public void ensureExecutedAsync(Runnable command) { try { delegate.execute(command); } catch (RejectedExecutionException t) { errorHandler.accept(command, t); ASYNC_POOL.execute(command); } } void shutdown() { if (delegate instanceof ExecutorService service) { service.shutdown(); } } } // We maintain a list of pending requests that will be aborted if ever // the selector manager thread exists abnormally. // The request, its id, and its completable future, are stored in a record-like // PendingRequest object added to the pending requests set (pendingRequests). // // When the request's cf completes, either normally or abnormally, a dependent action // will remove the PendingRequest object from the pending requests set. // If the SelectorManager threads exits abnormally, all pending requests in the // pending requests set will be completed exceptionally, in the ASYNC_POOL. // // HttpClientImpl::registerPending(id, req, cf, client) is called from sendAsync // to register the pending request in the pending requests set before returning // the completable future to the caller. // // HttpClientImpl::abortPendingRequests(client, Throwable reason) is called from // the SelectorManager when a throwable is caught just before exiting. // // A dependent action is registered with the pending request's cf to make sure // that the pending request will be removed from the pending requests set if, // or after, the cf is completed. // private final Set pendingRequests; // an id to ensure total order of pending request objects private final AtomicLong pendingRequestId = new AtomicLong(); private static final class PendingRequest implements Comparable { final long id; final HttpRequest request; final CompletableFuture cf; final HttpClientImpl client; final MultiExchange mex; Object ref; private PendingRequest(long id, HttpRequest request, CompletableFuture cf, MultiExchange mex, HttpClientImpl client) { this.id = id; this.request = request; this.cf = cf; this.mex = mex; this.client = client; } public void abort(Throwable t) { try { if (client.isSelectorThread()) { var done = cf.exceptionally((e) -> null); ASYNC_POOL.execute(() -> completeExceptionally(t)); // special case for when this method is called in the SelectorManager thread: // we want to wait until all futures are completed before proceeding to // shutdown. This ensures that the caller receive the actual `reason` // and not something like "HTTP/2 client closed"... done.join(); } else { cf.completeExceptionally(t); } } finally { mex.cancel(Utils.getIOException(t)); } } private void completeExceptionally(Throwable t) { if (client.debug.on()) { client.debug.log("aborting %s with %s", this, t); } try { cf.completeExceptionally(t); } catch (Throwable e) { client.debug.log("Failed to complete cf for [%s]: %s", this, e); } } @Override public int compareTo(PendingRequest o) { if (o == null) return 1; return Long.compare(id, o.id); } public String toString() { return id + ": " + request.toString(); } } static CompletableFuture registerPending(PendingRequest pending, CompletableFuture res) { // shortcut if cf is already completed: no need to go through the trouble of // registering it if (pending.cf.isDone()) return res; var client = pending.client; var cf = pending.cf; var id = pending.id; boolean added = client.pendingRequests.add(pending); // this may immediately remove `pending` from the set is the cf is already completed var ref = res.whenComplete((r,t) -> client.pendingRequests.remove(pending)); pending.ref = ref; assert added : "request %d was already added".formatted(id); // should not happen, unless the selector manager has already // exited abnormally if (client.selmgr.isClosed()) { pending.abort(client.selmgr.selectorClosedException()); } return ref; } static void abortPendingRequests(HttpClientImpl client, Throwable reason) { reason = Utils.getCompletionCause(reason); if (client.debug.on()) { var msg = reason instanceof RejectedExecutionException ? reason.getClass() : reason; client.debug.log("aborting pending requests due to: %s", msg); } closeSubscribers(client, reason); var pendingRequests = client.pendingRequests; while (!pendingRequests.isEmpty()) { var pendings = pendingRequests.iterator(); while (pendings.hasNext()) { var pending = pendings.next(); try { pending.abort(reason); } finally { pendings.remove(); } } } } private final CookieHandler cookieHandler; private final Duration connectTimeout; private final Redirect followRedirects; private final ProxySelector userProxySelector; private final ProxySelector proxySelector; private final Authenticator authenticator; private final Version version; private final ConnectionPool connections; private final DelegatingExecutor delegatingExecutor; private final boolean isDefaultExecutor; private final SSLContext sslContext; private final SSLParameters sslParams; private final SelectorManager selmgr; private final FilterFactory filters; private final Http2ClientImpl client2; private final long id; private final String dbgTag; private final InetAddress localAddr; // The SSL DirectBuffer Supplier provides the ability to recycle // buffers used between the socket reader and the SSLEngine, or // more precisely between the SocketTube publisher and the // SSLFlowDelegate reader. private final SSLDirectBufferSupplier sslBufferSupplier = new SSLDirectBufferSupplier(this); // This reference is used to keep track of the facade HttpClient // that was returned to the application code. // It makes it possible to know when the application no longer // holds any reference to the HttpClient. // Unfortunately, this information is not enough to know when // to exit the SelectorManager thread. Because of the asynchronous // nature of the API, we also need to wait until all pending operations // have completed. private final WeakReference facadeRef; private final WeakReference implRef; private final ConcurrentSkipListSet openedConnections = new ConcurrentSkipListSet<>(HttpConnection.COMPARE_BY_ID); private final ConcurrentSkipListSet> subscribers = new ConcurrentSkipListSet<>(HttpBodySubscriberWrapper.COMPARE_BY_ID); // This counter keeps track of the number of operations pending // on the HttpClient. The SelectorManager thread will wait // until there are no longer any pending operations and the // facadeRef is cleared before exiting. // // The pendingOperationCount is incremented every time a send/sendAsync // operation is invoked on the HttpClient, and is decremented when // the HttpResponse object is returned to the user. // However, at this point, the body may not have been fully read yet. // This is the case when the response T is implemented as a streaming // subscriber (such as an InputStream). // // To take care of this issue the pendingOperationCount will additionally // be incremented/decremented in the following cases: // // 1. For HTTP/2 it is incremented when a stream is added to the // Http2Connection streams map, and decreased when the stream is removed // from the map. This should also take care of push promises. // 2. For WebSocket the count is increased when creating a // DetachedConnectionChannel for the socket, and decreased // when the channel is closed. // In addition, the HttpClient facade is passed to the WebSocket builder, // (instead of the client implementation delegate). // 3. For HTTP/1.1 the count is incremented before starting to parse the body // response, and decremented when the parser has reached the end of the // response body flow. // // This should ensure that the selector manager thread remains alive until // the response has been fully received or the web socket is closed. private final AtomicLong pendingOperationCount = new AtomicLong(); private final AtomicLong pendingWebSocketCount = new AtomicLong(); private final AtomicLong pendingHttpOperationsCount = new AtomicLong(); private final AtomicLong pendingHttpRequestCount = new AtomicLong(); private final AtomicLong pendingHttp2StreamCount = new AtomicLong(); private final AtomicLong pendingTCPConnectionCount = new AtomicLong(); private final AtomicLong pendingSubscribersCount = new AtomicLong(); private final AtomicBoolean isAlive = new AtomicBoolean(); private final AtomicBoolean isStarted = new AtomicBoolean(); private volatile boolean shutdownRequested; /** A Set of, deadline first, ordered timeout events. */ private final TreeSet timeouts; /** * This is a bit tricky: * 1. an HttpClientFacade has a final HttpClientImpl field. * 2. an HttpClientImpl has a final WeakReference field, * where the referent is the facade created for that instance. * 3. We cannot just create the HttpClientFacade in the HttpClientImpl * constructor, because it would be only weakly referenced and could * be GC'ed before we can return it. * The solution is to use an instance of SingleFacadeFactory which will * allow the caller of new HttpClientImpl(...) to retrieve the facade * after the HttpClientImpl has been created. */ private static final class SingleFacadeFactory { HttpClientFacade facade; HttpClientFacade createFacade(HttpClientImpl impl) { assert facade == null; return (facade = new HttpClientFacade(impl)); } } static HttpClientFacade create(HttpClientBuilderImpl builder) { SingleFacadeFactory facadeFactory = new SingleFacadeFactory(); HttpClientImpl impl = new HttpClientImpl(builder, facadeFactory); impl.start(); assert facadeFactory.facade != null; assert impl.facadeRef.get() == facadeFactory.facade; return facadeFactory.facade; } private HttpClientImpl(HttpClientBuilderImpl builder, SingleFacadeFactory facadeFactory) { id = CLIENT_IDS.incrementAndGet(); dbgTag = "HttpClientImpl(" + id +")"; localAddr = builder.localAddr; if (builder.sslContext == null) { try { sslContext = SSLContext.getDefault(); } catch (NoSuchAlgorithmException ex) { throw new UncheckedIOException(new IOException(ex)); } } else { sslContext = builder.sslContext; } Executor ex = builder.executor; if (ex == null) { ex = Executors.newCachedThreadPool(new DefaultThreadFactory(id)); isDefaultExecutor = true; } else { isDefaultExecutor = false; } pendingRequests = new ConcurrentSkipListSet<>(); delegatingExecutor = new DelegatingExecutor(this::isSelectorThread, ex, this::onSubmitFailure); facadeRef = new WeakReference<>(facadeFactory.createFacade(this)); implRef = new WeakReference<>(this); client2 = new Http2ClientImpl(this); cookieHandler = builder.cookieHandler; connectTimeout = builder.connectTimeout; followRedirects = builder.followRedirects == null ? Redirect.NEVER : builder.followRedirects; this.userProxySelector = builder.proxy; this.proxySelector = Optional.ofNullable(userProxySelector) .orElseGet(ProxySelector::getDefault); if (debug.on()) debug.log("proxySelector is %s (user-supplied=%s)", this.proxySelector, userProxySelector != null); authenticator = builder.authenticator; if (builder.version == null) { version = HttpClient.Version.HTTP_2; } else { version = builder.version; } if (builder.sslParams == null) { sslParams = getDefaultParams(sslContext); } else { sslParams = builder.sslParams; } connections = new ConnectionPool(id); connections.start(); timeouts = new TreeSet<>(); try { selmgr = new SelectorManager(this); } catch (IOException e) { // unlikely throw new UncheckedIOException(e); } selmgr.setDaemon(true); filters = new FilterFactory(); initFilters(); assert facadeRef.get() != null; } // called when the facade is GC'ed. // Just wakes up the selector to cleanup... void facadeCleanup() { SelectorManager selmgr = this.selmgr; if (selmgr != null) selmgr.wakeupSelector(); } void onSubmitFailure(Runnable command, Throwable failure) { selmgr.abort(failure); } private void start() { try { selmgr.start(); } catch (Throwable t) { isStarted.set(true); throw t; } } // Called from the SelectorManager thread, just before exiting. // Clears the HTTP/1.1 and HTTP/2 cache, ensuring that the connections // that may be still lingering there are properly closed (and their // possibly still opened SocketChannel released). private void stop() { // Clears HTTP/1.1 cache and close its connections connections.stop(); // Clears HTTP/2 cache and close its connections. client2.stop(); // make sure all subscribers are completed closeSubscribers(); // close TCP connection if any are still opened openedConnections.forEach(this::closeConnection); // shutdown the executor if needed if (isDefaultExecutor) delegatingExecutor.shutdown(); } private void closeSubscribers() { if (subscribers.isEmpty()) return; IOException io = selmgr.selectorClosedException(); closeSubscribers(this, io); } private static void closeSubscribers(HttpClientImpl client, Throwable t) { client.subscribers.forEach(s -> s.onError(t)); } /** * Adds the given subscriber to the subscribers list, or call * its {@linkplain HttpBodySubscriberWrapper#onError onError} * method if the client is shutting down. * @param subscriber the subscriber * @return true if the subscriber was added to the list. */ public boolean registerSubscriber(HttpBodySubscriberWrapper subscriber) { if (!selmgr.isClosed()) { selmgr.lock(); try { if (!selmgr.isClosed()) { if (subscribers.add(subscriber)) { long count = pendingSubscribersCount.incrementAndGet(); if (debug.on()) { debug.log("body subscriber registered: " + count); } return true; } return false; } } finally { selmgr.unlock(); } } subscriber.onError(selmgr.selectorClosedException()); return false; } /** * Remove the given subscriber from the subscribers list. * @param subscriber the subscriber * @return true if the subscriber was found and removed from the list. */ public boolean unregisterSubscriber(HttpBodySubscriberWrapper subscriber) { if (subscribers.remove(subscriber)) { long count = pendingSubscribersCount.decrementAndGet(); if (debug.on()) { debug.log("body subscriber unregistered: " + count); } return true; } return false; } private void closeConnection(HttpConnection conn) { try { conn.close(); } catch (Throwable e) { if (Log.channel()) { Log.logChannel("Failed to close connection: " + e); } } } @Override public void shutdown() { shutdownRequested = true; selmgr.wakeupSelector(); } @Override public void shutdownNow() { shutdown(); selmgr.abort(new IOException("shutdownNow")); } @Override public boolean awaitTermination(Duration duration) throws InterruptedException { // Implicit NPE will be thrown if duration is null return selmgr.join(duration); } @Override public boolean isTerminated() { return isStarted.get() && !isAlive.get(); } private static SSLParameters getDefaultParams(SSLContext ctx) { SSLParameters params = ctx.getDefaultSSLParameters(); return params; } // Returns the facade that was returned to the application code. // May be null if that facade is no longer referenced. final HttpClientFacade facade() { return facadeRef.get(); } public long newConnectionId() { return CONNECTION_IDS.incrementAndGet(); } // Increments the pendingTCPConnectionCount public void connectionOpened(PlainHttpConnection plainHttpConnection) { if (!finished()) { if (openedConnections.add(plainHttpConnection)) { pendingTCPConnectionCount.incrementAndGet(); } } else closeConnection(plainHttpConnection); } // Decrements the pendingTCPConnectionCount public void connectionClosed(PlainHttpConnection plainHttpConnection) { if (openedConnections.remove(plainHttpConnection)) { pendingTCPConnectionCount.decrementAndGet(); } } // Increments the pendingHttpRequestCount and pendingOperationCount. final long requestReference() { pendingHttpRequestCount.incrementAndGet(); return reference(); } // Decrements the pendingHttpRequestCount and pendingOperationCount. final long requestUnreference() { pendingHttpRequestCount.decrementAndGet(); return unreference(); } // Increments the pendingHttpOperationsCount and pendingOperationCount. final long reference() { pendingHttpOperationsCount.incrementAndGet(); return pendingOperationCount.incrementAndGet(); } // Decrements the pendingHttpOperationsCount and pendingOperationCount. final long unreference() { final long count = pendingOperationCount.decrementAndGet(); final long httpCount = pendingHttpOperationsCount.decrementAndGet(); final long http2Count = pendingHttp2StreamCount.get(); final long webSocketCount = pendingWebSocketCount.get(); if (count == 0 && (facadeRef.refersTo(null) || shutdownRequested)) { selmgr.wakeupSelector(); } assert httpCount >= 0 : "count of HTTP/1.1 operations < 0"; assert http2Count >= 0 : "count of HTTP/2 operations < 0"; assert webSocketCount >= 0 : "count of WS operations < 0"; assert count >= 0 : "count of pending operations < 0"; return count; } // Increments the pendingHttp2StreamCount and pendingOperationCount. final long streamReference() { pendingHttp2StreamCount.incrementAndGet(); return pendingOperationCount.incrementAndGet(); } // Decrements the pendingHttp2StreamCount and pendingOperationCount. final long streamUnreference() { final long count = pendingOperationCount.decrementAndGet(); final long http2Count = pendingHttp2StreamCount.decrementAndGet(); final long httpCount = pendingHttpOperationsCount.get(); final long webSocketCount = pendingWebSocketCount.get(); if (count == 0 && (facadeRef.refersTo(null) || shutdownRequested)) { selmgr.wakeupSelector(); } assert httpCount >= 0 : "count of HTTP/1.1 operations < 0"; assert http2Count >= 0 : "count of HTTP/2 operations < 0"; assert webSocketCount >= 0 : "count of WS operations < 0"; assert count >= 0 : "count of pending operations < 0"; return count; } // Increments the pendingWebSocketCount and pendingOperationCount. final long webSocketOpen() { pendingWebSocketCount.incrementAndGet(); return pendingOperationCount.incrementAndGet(); } // Decrements the pendingWebSocketCount and pendingOperationCount. final long webSocketClose() { final long count = pendingOperationCount.decrementAndGet(); final long webSocketCount = pendingWebSocketCount.decrementAndGet(); final long httpCount = pendingHttpOperationsCount.get(); final long http2Count = pendingHttp2StreamCount.get(); if (count == 0 && (facadeRef.refersTo(null) || shutdownRequested)) { selmgr.wakeupSelector(); } assert httpCount >= 0 : "count of HTTP/1.1 operations < 0"; assert http2Count >= 0 : "count of HTTP/2 operations < 0"; assert webSocketCount >= 0 : "count of WS operations < 0"; assert count >= 0 : "count of pending operations < 0"; return count; } // Returns the pendingOperationCount. // Incremented with any operation, whether it's HTTP/1.1, HTTP/2, or WebSocket final long referenceCount() { return pendingOperationCount.get(); } // Trackers are used in test to verify that an instance of // HttpClient has shutdown correctly, and that all operations // have terminated. static final class HttpClientTracker implements Tracker { final AtomicLong requestCount; final AtomicLong httpCount; final AtomicLong http2Count; final AtomicLong websocketCount; final AtomicLong operationsCount; final AtomicLong connnectionsCount; final AtomicLong subscribersCount; final Reference reference; final Reference implRef; final AtomicBoolean isAlive; final AtomicBoolean isStarted; final String name; HttpClientTracker(AtomicLong request, AtomicLong http, AtomicLong http2, AtomicLong ws, AtomicLong ops, AtomicLong conns, AtomicLong subscribers, Reference ref, Reference implRef, AtomicBoolean isAlive, AtomicBoolean isStarted, String name) { this.requestCount = request; this.httpCount = http; this.http2Count = http2; this.websocketCount = ws; this.operationsCount = ops; this.connnectionsCount = conns; this.subscribersCount = subscribers; this.reference = ref; this.implRef = implRef; this.isAlive = isAlive; this.isStarted = isStarted; this.name = name; } @Override public long getOutstandingSubscribers() { return subscribersCount.get(); } @Override public long getOutstandingOperations() { return operationsCount.get(); } @Override public long getOutstandingHttpRequests() { return requestCount.get(); } @Override public long getOutstandingTcpConnections() { return connnectionsCount.get();} @Override public long getOutstandingHttpOperations() { return httpCount.get(); } @Override public long getOutstandingHttp2Streams() { return http2Count.get(); } @Override public long getOutstandingWebSocketOperations() { return websocketCount.get(); } @Override public boolean isFacadeReferenced() { return !reference.refersTo(null); } public boolean isImplementationReferenced() { return !implRef.refersTo(null); } // The selector is considered alive if it's not yet started @Override public boolean isSelectorAlive() { return isAlive.get() || !isStarted.get(); } @Override public String getName() { return name; } } public Tracker getOperationsTracker() { return new HttpClientTracker( pendingHttpRequestCount, pendingHttpOperationsCount, pendingHttp2StreamCount, pendingWebSocketCount, pendingOperationCount, pendingTCPConnectionCount, pendingSubscribersCount, facadeRef, implRef, isAlive, isStarted, dbgTag); } // Called by the SelectorManager thread to figure out whether it's time // to terminate. boolean finished() { if (referenceCount() > 0) return false; if (shutdownRequested) { synchronized (this) { // avoid race with sendAsync if (referenceCount() == 0) return true; } } return !isReferenced(); } private boolean isReferenced() { return !facadeRef.refersTo(null) || referenceCount() > 0; } /** * Wait for activity on given exchange. * The following occurs in the SelectorManager thread. * * 1) add to selector * 2) If selector fires for this exchange then * call AsyncEvent.handle() * * If exchange needs to change interest ops, then call registerEvent() again. */ void registerEvent(AsyncEvent exchange) throws IOException { selmgr.register(exchange); } /** * Allows an AsyncEvent to modify its interestOps. * @param event The modified event. */ void eventUpdated(AsyncEvent event) throws ClosedChannelException { assert !(event instanceof AsyncTriggerEvent); selmgr.eventUpdated(event); } boolean isSelectorThread() { return Thread.currentThread() == selmgr; } boolean isSelectorClosed() { return selmgr.isClosed(); } IOException selectorClosedException() { return selmgr.selectorClosedException(); } Http2ClientImpl client2() { return client2; } private void debugCompleted(String tag, long startNanos, HttpRequest req) { if (debugelapsed.on()) { debugelapsed.log(tag + " elapsed " + (System.nanoTime() - startNanos)/1000_000L + " millis for " + req.method() + " to " + req.uri()); } } @Override public HttpResponse send(HttpRequest req, BodyHandler responseHandler) throws IOException, InterruptedException { CompletableFuture> cf = null; // if the thread is already interrupted no need to go further. // cf.get() would throw anyway. if (Thread.interrupted()) throw new InterruptedException(); try { cf = sendAsync(req, responseHandler, null, null); return cf.get(); } catch (InterruptedException ie) { if (cf != null) { cf.cancel(true); } throw ie; } catch (ExecutionException e) { final Throwable throwable = e.getCause(); final String msg = throwable.getMessage(); if (throwable instanceof IllegalArgumentException) { throw new IllegalArgumentException(msg, throwable); } else if (throwable instanceof SecurityException) { throw new SecurityException(msg, throwable); } else if (throwable instanceof HttpConnectTimeoutException) { HttpConnectTimeoutException hcte = new HttpConnectTimeoutException(msg); hcte.initCause(throwable); throw hcte; } else if (throwable instanceof HttpTimeoutException) { throw new HttpTimeoutException(msg); } else if (throwable instanceof ConnectException) { ConnectException ce = new ConnectException(msg); ce.initCause(throwable); throw ce; } else if (throwable instanceof SSLHandshakeException) { // special case for SSLHandshakeException throw new SSLHandshakeException(msg, throwable); } else if (throwable instanceof SSLException) { // any other SSLException is wrapped in a plain // SSLException throw new SSLException(msg, throwable); } else if (throwable instanceof ProtocolException) { ProtocolException pe = new ProtocolException(msg); pe.initCause(throwable); throw pe; } else if (throwable instanceof IOException) { throw new IOException(msg, throwable); } else { throw new IOException(msg, throwable); } } } private static final Executor ASYNC_POOL = new CompletableFuture().defaultExecutor(); @Override public CompletableFuture> sendAsync(HttpRequest userRequest, BodyHandler responseHandler) { return sendAsync(userRequest, responseHandler, null); } @Override public CompletableFuture> sendAsync(HttpRequest userRequest, BodyHandler responseHandler, PushPromiseHandler pushPromiseHandler) { return sendAsync(userRequest, responseHandler, pushPromiseHandler, delegatingExecutor.delegate); } private CompletableFuture> sendAsync(HttpRequest userRequest, BodyHandler responseHandler, PushPromiseHandler pushPromiseHandler, Executor exchangeExecutor) { Objects.requireNonNull(userRequest); Objects.requireNonNull(responseHandler); if (shutdownRequested) { return MinimalFuture.failedFuture(new IOException("closed")); } // should not happen, unless the selector manager has // exited abnormally if (selmgr.isClosed()) { return MinimalFuture.failedFuture(selmgr.selectorClosedException()); } // Clone the possibly untrusted HttpRequest HttpRequestImpl requestImpl = new HttpRequestImpl(userRequest, proxySelector); if (requestImpl.method().equals("CONNECT")) throw new IllegalArgumentException("Unsupported method CONNECT"); long id = pendingRequestId.incrementAndGet(); long start = DEBUGELAPSED ? System.nanoTime() : 0; synchronized (this) { if (shutdownRequested) { return MinimalFuture.failedFuture(new IOException("closed")); } requestReference(); } try { if (debugelapsed.on()) debugelapsed.log("ClientImpl (async) send %s", userRequest); // When using sendAsync(...) we explicitly pass the // executor's delegate as exchange executor to force // 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 mex = new MultiExchange<>(userRequest, requestImpl, this, responseHandler, pushPromiseHandler); CompletableFuture> mexCf = mex.responseAsync(executor); CompletableFuture> res = mexCf.whenComplete((b,t) -> requestUnreference()); if (DEBUGELAPSED) { res = res.whenComplete( (b,t) -> debugCompleted("ClientImpl (async)", start, userRequest)); } // The mexCf is the Cf we need to abort if the SelectorManager thread // is aborted. PendingRequest pending = new PendingRequest(id, requestImpl, mexCf, mex, this); res = registerPending(pending, res); if (exchangeExecutor != null) { // makes sure that any dependent actions happen in the CF default // executor. This is only needed for sendAsync(...), when // exchangeExecutor is non-null. return res.isDone() ? res : res.whenCompleteAsync((r, t) -> { /* do nothing */}, ASYNC_POOL); } else { // make a defensive copy that can be safely canceled // by the caller return res.isDone() ? res : res.copy(); } } catch (Throwable t) { requestUnreference(); debugCompleted("ClientImpl (async)", start, userRequest); throw t; } } // Main loop for this client's selector private static final class SelectorManager extends Thread { // For testing purposes we have an internal System property that // can control the frequency at which the selector manager will wake // up when there are no pending operations. // Increasing the frequency (shorter delays) might allow the selector // to observe that the facade is no longer referenced and might allow // the selector thread to terminate more timely - for when nothing is // ongoing it will only check for that condition every NODEADLINE ms. // To avoid misuse of the property, the delay that can be specified // is comprised between [MIN_NODEADLINE, MAX_NODEADLINE], and its default // value if unspecified (or <= 0) is DEF_NODEADLINE = 3000ms // The property is -Djdk.internal.httpclient.selectorTimeout= private static final int MIN_NODEADLINE = 1000; // ms private static final int MAX_NODEADLINE = 1000 * 1200; // ms private static final int DEF_NODEADLINE = 3000; // ms private static final long NODEADLINE; // default is DEF_NODEADLINE ms static { // ensure NODEADLINE is initialized with some valid value. long deadline = Utils.getIntegerProperty( "jdk.internal.httpclient.selectorTimeout", DEF_NODEADLINE); // millis if (deadline <= 0) deadline = DEF_NODEADLINE; deadline = Math.max(deadline, MIN_NODEADLINE); NODEADLINE = Math.min(deadline, MAX_NODEADLINE); } private final Selector selector; private volatile boolean closed; private final List registrations; private final List deregistrations; private final Logger debug; private final Logger debugtimeout; private final HttpClientImpl owner; private final ConnectionPool pool; private final AtomicReference errorRef = new AtomicReference<>(); private final ReentrantLock lock = new ReentrantLock(); SelectorManager(HttpClientImpl ref) throws IOException { super(null, null, "HttpClient-" + ref.id + "-SelectorManager", 0, false); owner = ref; debug = ref.debug; debugtimeout = ref.debugtimeout; pool = ref.connectionPool(); registrations = new ArrayList<>(); deregistrations = new ArrayList<>(); selector = Selector.open(); } IOException selectorClosedException() { var io = new IOException("selector manager closed"); var cause = errorRef.get(); if (cause != null) { io.initCause(cause); } return io; } void eventUpdated(AsyncEvent e) throws ClosedChannelException { if (Thread.currentThread() == this) { SelectionKey key = e.channel().keyFor(selector); if (key != null && key.isValid()) { SelectorAttachment sa = (SelectorAttachment) key.attachment(); sa.register(e); } else if (e.interestOps() != 0) { // We don't care about paused events. // These are actually handled by // SelectorAttachment::resetInterestOps later on. // But if we reach here when trying to resume an // event then it's better to fail fast. if (debug.on()) debug.log("No key for channel"); e.abort(new IOException("No key for channel")); } } else { register(e); } } // This returns immediately. So caller not allowed to send/receive // on connection. void register(AsyncEvent e) { var closed = this.closed; if (!closed) { lock.lock(); try { closed = this.closed; if (!closed) { registrations.add(e); } } finally { lock.unlock(); } } if (closed) { e.abort(selectorClosedException()); } else { selector.wakeup(); } } void wakeupSelector() { selector.wakeup(); } void abort(Throwable t) { boolean closed = this.closed; errorRef.compareAndSet(null, t); if (debug.on()) { debug.log("aborting selector manager(closed=%s): " + t, closed); } t = errorRef.get(); boolean inSelectorThread = owner.isSelectorThread(); if (!inSelectorThread) { // abort anything pending, then close abortPendingRequests(owner, t); } Set keys = new HashSet<>(); Set toAbort = new HashSet<>(); lock.lock(); try { if (closed = this.closed) return; this.closed = true; try { keys.addAll(selector.keys()); } catch (ClosedSelectorException ce) { // OK - nothing to do... } toAbort.addAll(this.registrations); toAbort.addAll(this.deregistrations); this.registrations.clear(); this.deregistrations.clear(); } finally { lock.unlock(); } // double check after closing abortPendingRequests(owner, t); IOException io = toAbort.isEmpty() ? null : selectorClosedException(); for (AsyncEvent e : toAbort) { try { e.abort(io); } catch (Throwable x) { debug.log("Failed to abort event: " + x); } } if (!inSelectorThread) selector.wakeup(); } // Only called by the selector manager thread private void shutdown() { try { lock.lock(); try { Log.logTrace("{0}: shutting down", getName()); if (debug.on()) debug.log("SelectorManager shutting down"); closed = true; selector.close(); } finally { lock.unlock(); } } catch (IOException ignored) { } finally { owner.stop(); } } boolean isClosed() { return closed; } @Override public void run() { List> errorList = new ArrayList<>(); List readyList = new ArrayList<>(); List resetList = new ArrayList<>(); owner.isAlive.set(true); // goes back to false when run exits owner.isStarted.set(true); // never goes back to false try { if (Log.channel()) Log.logChannel(getName() + ": starting"); while (!Thread.currentThread().isInterrupted() && !closed) { lock.lock(); try { assert errorList.isEmpty(); assert readyList.isEmpty(); assert resetList.isEmpty(); for (AsyncTriggerEvent event : deregistrations) { event.handle(); } deregistrations.clear(); for (AsyncEvent event : registrations) { if (event instanceof AsyncTriggerEvent) { readyList.add(event); continue; } SelectableChannel chan = event.channel(); SelectionKey key = null; try { key = chan.keyFor(selector); SelectorAttachment sa; if (key == null || !key.isValid()) { if (key != null) { // key is canceled. // invoke selectNow() to purge it // before registering the new event. selector.selectNow(); } sa = new SelectorAttachment(chan, selector); } else { sa = (SelectorAttachment) key.attachment(); } // may throw IOE if channel closed: that's OK sa.register(event); if (!chan.isOpen()) { throw new ClosedChannelException(); } } catch (IOException 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.add(new Pair<>(event, e)); if (key != null) { key.cancel(); selector.selectNow(); } } } registrations.clear(); selector.selectedKeys().clear(); } finally { lock.unlock(); } for (AsyncEvent event : readyList) { assert event instanceof AsyncTriggerEvent; event.handle(); } readyList.clear(); for (Pair error : errorList) { // an IOException was raised and the channel closed. handleEvent(error.first, error.second); } errorList.clear(); // Check whether client is still alive, and if not, // gracefully stop this thread if (owner.finished()) { Log.logTrace("{0}: {1}", getName(), "HttpClient finished. Exiting..."); return; } // Timeouts will have milliseconds granularity. It is important // to handle them in a timely fashion. long nextTimeout = owner.purgeTimeoutsAndReturnNextDeadline(); if (debugtimeout.on()) debugtimeout.log("next timeout: %d", nextTimeout); // Keep-alive have seconds granularity. It's not really an // issue if we keep connections linger a bit more in the keep // alive cache. long nextExpiry = pool.purgeExpiredConnectionsAndReturnNextDeadline(); if (debugtimeout.on()) debugtimeout.log("next expired: %d", nextExpiry); assert nextTimeout >= 0; assert nextExpiry >= 0; // Don't wait for ever as it might prevent the thread to // stop gracefully. millis will be 0 if no deadline was found. if (nextTimeout <= 0) nextTimeout = NODEADLINE; // Clip nextExpiry at NODEADLINE limit. The default // keep alive is 1200 seconds (half an hour) - we don't // want to wait that long. if (nextExpiry <= 0) nextExpiry = NODEADLINE; else nextExpiry = Math.min(NODEADLINE, nextExpiry); // takes the least of the two. long millis = Math.min(nextExpiry, nextTimeout); if (debugtimeout.on()) debugtimeout.log("Next deadline is %d", (millis == 0 ? NODEADLINE : millis)); //debugPrint(selector); 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.finished()) { Log.logTrace("{0}: {1}", getName(), "HttpClient finished. Exiting..."); return; } owner.purgeTimeoutsAndReturnNextDeadline(); continue; } Set keys = selector.selectedKeys(); assert errorList.isEmpty(); for (SelectionKey key : keys) { SelectorAttachment sa = (SelectorAttachment) key.attachment(); if (!key.isValid()) { IOException ex = sa.chan.isOpen() ? new IOException("Invalid key") : new ClosedChannelException(); sa.pending.forEach(e -> errorList.add(new Pair<>(e,ex))); sa.pending.clear(); continue; } int eventsOccurred; try { eventsOccurred = key.readyOps(); } catch (CancelledKeyException ex) { IOException io = Utils.getIOException(ex); sa.pending.forEach(e -> errorList.add(new Pair<>(e,io))); sa.pending.clear(); continue; } sa.events(eventsOccurred).forEach(readyList::add); resetList.add(() -> sa.resetInterestOps(eventsOccurred)); } selector.selectNow(); // complete cancellation selector.selectedKeys().clear(); // handle selected events IOException ioe = closed ? selectorClosedException() : null; readyList.forEach((e) -> handleEvent(e, ioe)); readyList.clear(); // handle errors (closed channels etc...) errorList.forEach((p) -> handleEvent(p.first, p.second)); errorList.clear(); // reset interest ops for selected channels resetList.forEach(r -> r.run()); resetList.clear(); } } catch (Throwable e) { errorRef.compareAndSet(null, e); if (!closed) { closed = true; // set closed early so that new requests are rejected // This terminates thread. So, better just print stack trace String err = Utils.stackTrace(e); Log.logError("{0}: {1}: {2}", getName(), "HttpClientImpl shutting down due to fatal error", err); } abortPendingRequests(owner, selectorClosedException()); 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"); try { shutdown(); } finally { owner.isAlive.set(false); } } } // void debugPrint(Selector selector) { // System.err.println("Selector: debugprint start"); // Set keys = selector.keys(); // for (SelectionKey key : keys) { // SelectableChannel c = key.channel(); // int ops = key.interestOps(); // System.err.printf("selector chan:%s ops:%d\n", c, ops); // } // System.err.println("Selector: debugprint end"); // } /** Handles the given event. The given ioe may be null. */ void handleEvent(AsyncEvent event, IOException ioe) { if (ioe == null && closed) { ioe = selectorClosedException(); } if (ioe != null) { event.abort(ioe); } else { event.handle(); } } void lock() { lock.lock(); } void unlock() { lock.unlock(); } } final String debugInterestOps(SelectableChannel channel) { try { SelectionKey key = channel.keyFor(selmgr.selector); if (key == null) return "channel not registered with selector"; String keyInterestOps = key.isValid() ? "key.interestOps=" + Utils.interestOps(key) : "invalid key"; return String.format("channel registered with selector, %s, sa.interestOps=%s", keyInterestOps, Utils.describeOps(((SelectorAttachment)key.attachment()).interestOps)); } catch (Throwable t) { return String.valueOf(t); } } /** * Tracks multiple user level registrations associated with one NIO * registration (SelectionKey). In this implementation, registrations * are one-off and when an event is posted the registration is cancelled * until explicitly registered again. * *

No external synchronization required as this class is only used * by the SelectorManager thread. One of these objects required per * connection. */ private static class SelectorAttachment { private final SelectableChannel chan; private final Selector selector; private final Set pending; private static final Logger debug = Utils.getDebugLogger("SelectorAttachment"::toString, Utils.DEBUG); private int interestOps; SelectorAttachment(SelectableChannel chan, Selector selector) { this.pending = new HashSet<>(); this.chan = chan; this.selector = selector; } void register(AsyncEvent e) throws ClosedChannelException { int newOps = e.interestOps(); // re register interest if we are not already interested // in the event. If the event is paused, then the pause will // be taken into account later when resetInterestOps is called. boolean reRegister = (interestOps & newOps) != newOps; interestOps |= newOps; pending.add(e); if (debug.on()) debug.log("Registering %s for %s (%s)", e, Utils.describeOps(newOps), reRegister); if (reRegister) { // first time registration happens here also try { chan.register(selector, interestOps, this); } catch (Throwable x) { abortPending(x); } } else if (!chan.isOpen()) { abortPending(new ClosedChannelException()); } } /** * Returns a Stream containing only events that are * registered with the given {@code interestOps}. */ Stream events(int interestOps) { return pending.stream() .filter(ev -> (ev.interestOps() & interestOps) != 0); } /** * Removes any events with the given {@code interestOps}, and if no * events remaining, cancels the associated SelectionKey. */ void resetInterestOps(int interestOps) { int newOps = 0; Iterator itr = pending.iterator(); while (itr.hasNext()) { AsyncEvent event = itr.next(); int evops = event.interestOps(); if (event.repeating()) { newOps |= evops; continue; } if ((evops & interestOps) != 0) { itr.remove(); } else { newOps |= evops; } } this.interestOps = newOps; SelectionKey key = chan.keyFor(selector); if (newOps == 0 && key != null && pending.isEmpty()) { key.cancel(); } else { try { if (key == null || !key.isValid()) { throw new CancelledKeyException(); } key.interestOps(newOps); // double check after if (!chan.isOpen()) { abortPending(new ClosedChannelException()); return; } assert key.interestOps() == newOps; } catch (CancelledKeyException x) { // channel may have been closed if (debug.on()) debug.log("key cancelled for " + chan); abortPending(x); } } } void abortPending(Throwable x) { if (!pending.isEmpty()) { AsyncEvent[] evts = pending.toArray(new AsyncEvent[0]); pending.clear(); IOException io = Utils.getIOException(x); for (AsyncEvent event : evts) { event.abort(io); } } } } /*package-private*/ SSLContext theSSLContext() { return sslContext; } @Override public SSLContext sslContext() { return sslContext; } @Override public SSLParameters sslParameters() { return Utils.copySSLParameters(sslParams); } @Override public Optional authenticator() { return Optional.ofNullable(authenticator); } /*package-private*/ final DelegatingExecutor theExecutor() { return delegatingExecutor; } @Override public final Optional executor() { return isDefaultExecutor ? Optional.empty() : Optional.of(delegatingExecutor.delegate()); } ConnectionPool connectionPool() { return connections; } @Override public Redirect followRedirects() { return followRedirects; } @Override public Optional cookieHandler() { return Optional.ofNullable(cookieHandler); } @Override public Optional connectTimeout() { return Optional.ofNullable(connectTimeout); } Optional idleConnectionTimeout() { return Optional.ofNullable(getIdleConnectionTimeout()); } @Override public Optional proxy() { return Optional.ofNullable(userProxySelector); } // Return the effective proxy that this client uses. ProxySelector proxySelector() { return proxySelector; } @Override public WebSocket.Builder newWebSocketBuilder() { // Make sure to pass the HttpClientFacade to the WebSocket builder. // This will ensure that the facade is not released before the // WebSocket has been created, at which point the pendingOperationCount // will have been incremented by the RawChannelTube. // See RawChannelTube. return new BuilderImpl(this.facade(), proxySelector); } @Override public Version version() { return version; } InetAddress localAddress() { return localAddr; } String dbgString() { return dbgTag; } @Override public String toString() { // Used by tests to get the client's id and compute the // name of the SelectorManager thread. return super.toString() + ("(" + id + ")"); } private void initFilters() { addFilter(AuthenticationFilter.class); addFilter(RedirectFilter.class); if (this.cookieHandler != null) { addFilter(CookieFilter.class); } } private void addFilter(Class f) { filters.addFilter(f); } final List filterChain() { return filters.getFilterChain(); } // Timer controls. // Timers are implemented through timed Selector.select() calls. void registerTimer(TimeoutEvent event) { Log.logTrace("Registering timer {0}", event); synchronized (this) { timeouts.add(event); selmgr.wakeupSelector(); } } void cancelTimer(TimeoutEvent event) { Log.logTrace("Canceling timer {0}", event); synchronized (this) { timeouts.remove(event); } } /** * Purges ( handles ) timer events that have passed their deadline, and * returns the amount of time, in milliseconds, until the next earliest * event. A return value of 0 means that there are no events. */ private long purgeTimeoutsAndReturnNextDeadline() { long diff = 0L; List toHandle = null; int remaining = 0; // enter critical section to retrieve the timeout event to handle synchronized (this) { if (timeouts.isEmpty()) return 0L; Deadline now = TimeSource.now(); Iterator itr = timeouts.iterator(); while (itr.hasNext()) { TimeoutEvent event = itr.next(); diff = now.until(event.deadline(), ChronoUnit.MILLIS); if (diff <= 0) { itr.remove(); toHandle = (toHandle == null) ? new ArrayList<>() : toHandle; toHandle.add(event); } else { break; } } remaining = timeouts.size(); } // can be useful for debugging if (toHandle != null && Log.trace()) { Log.logTrace("purgeTimeoutsAndReturnNextDeadline: handling " + toHandle.size() + " events, " + "remaining " + remaining + ", next deadline: " + (diff < 0 ? 0L : diff)); } // handle timeout events out of critical section if (toHandle != null) { Throwable failed = null; for (TimeoutEvent event : toHandle) { try { Log.logTrace("Firing timer {0}", event); event.handle(); } catch (Error | RuntimeException e) { // Not expected. Handle remaining events then throw... // If e is an OOME or SOE it might simply trigger a new // error from here - but in this case there's not much we // could do anyway. Just let it flow... if (failed == null) failed = e; else failed.addSuppressed(e); Log.logTrace("Failed to handle event {0}: {1}", event, e); } } if (failed instanceof Error) throw (Error) failed; if (failed instanceof RuntimeException) throw (RuntimeException) failed; } // return time to wait until next event. 0L if there's no more events. return diff < 0 ? 0L : diff; } // used for the connection window int getReceiveBufferSize() { return Utils.getIntegerNetProperty( "jdk.httpclient.receiveBufferSize", 0 // only set the size if > 0 ); } // used for testing int getSendBufferSize() { return Utils.getIntegerNetProperty( "jdk.httpclient.sendBufferSize", 0 // only set the size if > 0 ); } // Optimization for reading SSL encrypted data // -------------------------------------------- // Returns a BufferSupplier that can be used for reading // encrypted bytes of the channel. These buffers can then // be recycled by the SSLFlowDelegate::Reader after their // content has been copied in the SSLFlowDelegate::Reader // readBuf. // Because allocating, reading, copying, and recycling // all happen in the SelectorManager thread, // then this BufferSupplier can be shared between all // the SSL connections managed by this client. BufferSupplier getSSLBufferSupplier() { return sslBufferSupplier; } private Duration getIdleConnectionTimeout() { if (IDLE_CONNECTION_TIMEOUT >= 0) return Duration.ofSeconds(IDLE_CONNECTION_TIMEOUT); return null; } private static long getTimeoutProp(String prop, long def) { String s = Utils.getNetProperty(prop); try { if (s != null) { long timeoutVal = Long.parseLong(s); if (timeoutVal >= 0) return timeoutVal; } } catch (NumberFormatException ignored) { Log.logTrace("Invalid value set for " + prop + " property: " + ignored); } return def; } // An implementation of BufferSupplier that manages a pool of // maximum 3 direct byte buffers (SocketTube.MAX_BUFFERS) that // are used for reading encrypted bytes off the channel before // copying and subsequent unwrapping. private static final class SSLDirectBufferSupplier implements BufferSupplier { private static final int POOL_SIZE = SocketTube.MAX_BUFFERS; private final ByteBuffer[] pool = new ByteBuffer[POOL_SIZE]; private final HttpClientImpl client; private final Logger debug; private int tail, count; // no need for volatile: only accessed in SM thread. SSLDirectBufferSupplier(HttpClientImpl client) { this.client = Objects.requireNonNull(client); this.debug = client.debug; } // Gets a buffer from the pool, or allocates a new one if needed. @Override public ByteBuffer get() { assert client.isSelectorThread(); assert tail <= POOL_SIZE : "allocate tail is " + tail; ByteBuffer buf; if (tail == 0) { if (debug.on()) { // should not appear more than SocketTube.MAX_BUFFERS debug.log("ByteBuffer.allocateDirect(%d)", Utils.BUFSIZE); } assert count++ < POOL_SIZE : "trying to allocate more than " + POOL_SIZE + " buffers"; buf = ByteBuffer.allocateDirect(Utils.BUFSIZE); } else { assert tail > 0 : "non positive tail value: " + tail; tail--; buf = pool[tail]; pool[tail] = null; } assert buf.isDirect(); assert buf.position() == 0; assert buf.hasRemaining(); assert buf.limit() == Utils.BUFSIZE; assert tail < POOL_SIZE; assert tail >= 0; return buf; } // Returns the given buffer to the pool. @Override public void recycle(ByteBuffer buffer) { assert client.isSelectorThread(); assert buffer.isDirect(); assert !buffer.hasRemaining(); assert tail < POOL_SIZE : "recycle tail is " + tail; assert tail >= 0; buffer.position(0); buffer.limit(buffer.capacity()); // don't fail if assertions are off. we have asserted above. if (tail < POOL_SIZE) { pool[tail] = buffer; tail++; } assert tail <= POOL_SIZE; assert tail > 0; } } }