1 /*

     2  * Copyright (c) 2015, 2017, Oracle and/or its affiliates. All rights reserved.

     3  * DO NOT ALTER OR REMOVE COPYRIGHT NOTICES OR THIS FILE HEADER.

     4  *

     5  * This code is free software; you can redistribute it and/or modify it

     6  * under the terms of the GNU General Public License version 2 only, as

     7  * published by the Free Software Foundation.  Oracle designates this

     8  * particular file as subject to the "Classpath" exception as provided

     9  * by Oracle in the LICENSE file that accompanied this code.

    10  *

    11  * This code is distributed in the hope that it will be useful, but WITHOUT

    12  * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or

    13  * FITNESS FOR A PARTICULAR PURPOSE.  See the GNU General Public License

    14  * version 2 for more details (a copy is included in the LICENSE file that

    15  * accompanied this code).

    16  *

    17  * You should have received a copy of the GNU General Public License version

    18  * 2 along with this work; if not, write to the Free Software Foundation,

    19  * Inc., 51 Franklin St, Fifth Floor, Boston, MA 02110-1301 USA.

    20  *

    21  * Please contact Oracle, 500 Oracle Parkway, Redwood Shores, CA 94065 USA

    22  * or visit www.oracle.com if you need additional information or have any

    23  * questions.

    24  */

    25 

    26 package jdk.incubator.http;

    27 

    28 import javax.net.ssl.SSLContext;

    29 import javax.net.ssl.SSLParameters;

    30 import java.io.IOException;

    31 import java.lang.ref.WeakReference;

    32 import java.net.Authenticator;

    33 import java.net.CookieManager;

    34 import java.net.ProxySelector;

    35 import java.net.URI;

    36 import java.nio.channels.ClosedChannelException;

    37 import java.nio.channels.SelectableChannel;

    38 import java.nio.channels.SelectionKey;

    39 import java.nio.channels.Selector;

    40 import java.nio.channels.SocketChannel;

    41 import java.security.NoSuchAlgorithmException;

    42 import java.time.Instant;

    43 import java.time.temporal.ChronoUnit;

    44 import java.util.ArrayList;

    45 import java.util.Iterator;

    46 import java.util.List;

    47 import java.util.Optional;

    48 import java.util.Set;

    49 import java.util.TreeSet;

    50 import java.util.concurrent.CompletableFuture;

    51 import java.util.concurrent.Executor;

    52 import java.util.concurrent.Executors;

    53 import java.util.concurrent.ThreadFactory;

    54 import java.util.stream.Stream;

    55 import jdk.incubator.http.internal.common.Log;

    56 import jdk.incubator.http.internal.common.Utils;

    57 import jdk.incubator.http.internal.websocket.BuilderImpl;

    58 

    59 /**

    60  * Client implementation. Contains all configuration information and also

    61  * the selector manager thread which allows async events to be registered

    62  * and delivered when they occur. See AsyncEvent.

    63  */

    64 class HttpClientImpl extends HttpClient {

    65 

    66     // Define the default factory as a static inner class

    67     // that embeds all the necessary logic to avoid

    68     // the risk of using a lambda that might keep a reference on the

    69     // HttpClient instance from which it was created (helps with

    70     // heapdump analysis).

    71     private static final class DefaultThreadFactory implements ThreadFactory {

    72         private DefaultThreadFactory() {}

    73         @Override

    74         public Thread newThread(Runnable r) {

    75             Thread t = new Thread(null, r, "HttpClient_worker", 0, true);

    76             t.setDaemon(true);

    77             return t;

    78         }

    79         static final ThreadFactory INSTANCE = new DefaultThreadFactory();

    80     }

    81 

    82     private final CookieManager cookieManager;

    83     private final Redirect followRedirects;

    84     private final ProxySelector proxySelector;

    85     private final Authenticator authenticator;

    86     private final Version version;

    87     private final ConnectionPool connections;

    88     private final Executor executor;

    89     // Security parameters

    90     private final SSLContext sslContext;

    91     private final SSLParameters sslParams;

    92     private final SelectorManager selmgr;

    93     private final FilterFactory filters;

    94     private final Http2ClientImpl client2;

    95 

    96     /** A Set of, deadline first, ordered timeout events. */

    97     private final TreeSet<TimeoutEvent> timeouts;

    98 

    99     public static HttpClientImpl create(HttpClientBuilderImpl builder) {

   100         HttpClientImpl impl = new HttpClientImpl(builder);

   101         impl.start();

   102         return impl;

   103     }

   104 

   105     private HttpClientImpl(HttpClientBuilderImpl builder) {

   106         if (builder.sslContext == null) {

   107             try {

   108                 sslContext = SSLContext.getDefault();

   109             } catch (NoSuchAlgorithmException ex) {

   110                 throw new InternalError(ex);

   111             }

   112         } else {

   113             sslContext = builder.sslContext;

   114         }

   115         Executor ex = builder.executor;

   116         if (ex == null) {

   117             ex = Executors.newCachedThreadPool(DefaultThreadFactory.INSTANCE);

   118         } else {

   119             ex = builder.executor;

   120         }

   121         client2 = new Http2ClientImpl(this);

   122         executor = ex;

   123         cookieManager = builder.cookieManager;

   124         followRedirects = builder.followRedirects == null ?

   125                 Redirect.NEVER : builder.followRedirects;

   126         this.proxySelector = builder.proxy;

   127         authenticator = builder.authenticator;

   128         if (builder.version == null) {

   129             version = HttpClient.Version.HTTP_2;

   130         } else {

   131             version = builder.version;

   132         }

   133         if (builder.sslParams == null) {

   134             sslParams = getDefaultParams(sslContext);

   135         } else {

   136             sslParams = builder.sslParams;

   137         }

   138         connections = new ConnectionPool();

   139         connections.start();

   140         timeouts = new TreeSet<>();

   141         try {

   142             selmgr = new SelectorManager(this);

   143         } catch (IOException e) {

   144             // unlikely

   145             throw new InternalError(e);

   146         }

   147         selmgr.setDaemon(true);

   148         filters = new FilterFactory();

   149         initFilters();

   150     }

   151 

   152     private void start() {

   153         selmgr.start();

   154     }

   155 

   156     private static SSLParameters getDefaultParams(SSLContext ctx) {

   157         SSLParameters params = ctx.getSupportedSSLParameters();

   158         params.setProtocols(new String[]{"TLSv1.2"});

   159         return params;

   160     }

   161 

   162     /**

   163      * Wait for activity on given exchange (assuming blocking = false).

   164      * It's a no-op if blocking = true. In particular, the following occurs

   165      * in the SelectorManager thread.

   166      *

   167      *  1) mark the connection non-blocking

   168      *  2) add to selector

   169      *  3) If selector fires for this exchange then

   170      *  4)   - mark connection as blocking

   171      *  5)   - call AsyncEvent.handle()

   172      *

   173      * If exchange needs to block again, then call registerEvent() again

   174      */

   175     void registerEvent(AsyncEvent exchange) throws IOException {

   176         selmgr.register(exchange);

   177     }

   178 

   179     /**

   180      * Only used from RawChannel to disconnect the channel from

   181      * the selector

   182      */

   183     void cancelRegistration(SocketChannel s) {

   184         selmgr.cancel(s);

   185     }

   186 

   187 

   188     Http2ClientImpl client2() {

   189         return client2;

   190     }

   191 

   192     /*

   193     @Override

   194     public ByteBuffer getBuffer() {

   195         return pool.getBuffer();

   196     }

   197 

   198     // SSL buffers are larger. Manage separately

   199 

   200     int size = 16 * 1024;

   201 

   202     ByteBuffer getSSLBuffer() {

   203         return ByteBuffer.allocate(size);

   204     }

   205 

   206     /**

   207      * Return a new buffer that's a bit bigger than the given one

   208      *

   209      * @param buf

   210      * @return

   211      *

   212     ByteBuffer reallocSSLBuffer(ByteBuffer buf) {

   213         size = buf.capacity() * 12 / 10; // 20% bigger

   214         return ByteBuffer.allocate(size);

   215     }

   216 

   217     synchronized void returnSSLBuffer(ByteBuffer buf) {

   218         if (buf.capacity() >= size)

   219            sslBuffers.add(0, buf);

   220     }

   221 

   222     @Override

   223     public void returnBuffer(ByteBuffer buffer) {

   224         pool.returnBuffer(buffer);

   225     }

   226     */

   227 

   228     @Override

   229     public <T> HttpResponse<T>

   230     send(HttpRequest req, HttpResponse.BodyHandler<T> responseHandler)

   231         throws IOException, InterruptedException

   232     {

   233         MultiExchange<Void,T> mex = new MultiExchange<>(req, this, responseHandler);

   234         return mex.response();

   235     }

   236 

   237     @Override

   238     public <T> CompletableFuture<HttpResponse<T>>

   239     sendAsync(HttpRequest req, HttpResponse.BodyHandler<T> responseHandler)

   240     {

   241         MultiExchange<Void,T> mex = new MultiExchange<>(req, this, responseHandler);

   242         return mex.responseAsync()

   243                   .thenApply((HttpResponseImpl<T> b) -> (HttpResponse<T>) b);

   244     }

   245 

   246     @Override

   247     public <U, T> CompletableFuture<U>

   248     sendAsync(HttpRequest req, HttpResponse.MultiProcessor<U, T> responseHandler) {

   249         MultiExchange<U,T> mex = new MultiExchange<>(req, this, responseHandler);

   250         return mex.multiResponseAsync();

   251     }

   252 

   253     // new impl. Should get rid of above

   254     /*

   255     static class BufferPool implements BufferHandler {

   256 

   257         final LinkedList<ByteBuffer> freelist = new LinkedList<>();

   258 

   259         @Override

   260         public synchronized ByteBuffer getBuffer() {

   261             ByteBuffer buf;

   262 

   263             while (!freelist.isEmpty()) {

   264                 buf = freelist.removeFirst();

   265                 buf.clear();

   266                 return buf;

   267             }

   268             return ByteBuffer.allocate(BUFSIZE);

   269         }

   270 

   271         @Override

   272         public synchronized void returnBuffer(ByteBuffer buffer) {

   273             assert buffer.capacity() > 0;

   274             freelist.add(buffer);

   275         }

   276     }

   277 

   278     static BufferPool pool = new BufferPool();

   279 

   280     static BufferHandler pool() {

   281         return pool;

   282     }

   283 */

   284     // Main loop for this client's selector

   285     private final static class SelectorManager extends Thread {

   286 

   287         private static final long NODEADLINE = 3000L;

   288         private final Selector selector;

   289         private volatile boolean closed;

   290         private final List<AsyncEvent> readyList;

   291         private final List<AsyncEvent> registrations;

   292 

   293         // Uses a weak reference to the HttpClient owning this

   294         // selector: a strong reference prevents its garbage

   295         // collection while the thread is running.

   296         // We want the thread to exit gracefully when the

   297         // HttpClient that owns it gets GC'ed.

   298         WeakReference<HttpClientImpl> ownerRef;

   299 

   300         SelectorManager(HttpClientImpl ref) throws IOException {

   301             super(null, null, "SelectorManager", 0, false);

   302             ownerRef = new WeakReference<>(ref);

   303             readyList = new ArrayList<>();

   304             registrations = new ArrayList<>();

   305             selector = Selector.open();

   306         }

   307 

   308         // This returns immediately. So caller not allowed to send/receive

   309         // on connection.

   310 

   311         synchronized void register(AsyncEvent e) throws IOException {

   312             registrations.add(e);

   313             selector.wakeup();

   314         }

   315 

   316         synchronized void cancel(SocketChannel e) {

   317             SelectionKey key = e.keyFor(selector);

   318             if (key != null) {

   319                 key.cancel();

   320             }

   321             selector.wakeup();

   322         }

   323 

   324         void wakeupSelector() {

   325             selector.wakeup();

   326         }

   327 

   328         synchronized void shutdown() {

   329             closed = true;

   330             try {

   331                 selector.close();

   332             } catch (IOException ignored) { }

   333         }

   334 

   335         @Override

   336         public void run() {

   337             try {

   338                 while (!Thread.currentThread().isInterrupted()) {

   339                     HttpClientImpl client;

   340                     synchronized (this) {

   341                         for (AsyncEvent exchange : registrations) {

   342                             SelectableChannel c = exchange.channel();

   343                             try {

   344                                 c.configureBlocking(false);

   345                                 SelectionKey key = c.keyFor(selector);

   346                                 SelectorAttachment sa;

   347                                 if (key == null || !key.isValid()) {

   348                                     if (key != null) {

   349                                         // key is canceled.

   350                                         // invoke selectNow() to purge it

   351                                         // before registering the new event.

   352                                         selector.selectNow();

   353                                     }

   354                                     sa = new SelectorAttachment(c, selector);

   355                                 } else {

   356                                     sa = (SelectorAttachment) key.attachment();

   357                                 }

   358                                 sa.register(exchange);

   359                             } catch (IOException e) {

   360                                 Log.logError("HttpClientImpl: " + e);

   361                                 c.close();

   362                                 // let the exchange deal with it

   363                                 handleEvent(exchange);

   364                             }

   365                         }

   366                         registrations.clear();

   367                     }

   368 

   369                     // Check whether client is still alive, and if not,

   370                     // gracefully stop this thread

   371                     if ((client = ownerRef.get()) == null) {

   372                         Log.logTrace("HttpClient no longer referenced. Exiting...");

   373                         return;

   374                     }

   375                     long millis = client.purgeTimeoutsAndReturnNextDeadline();

   376                     client = null; // don't hold onto the client ref

   377 

   378                     //debugPrint(selector);

   379                     // Don't wait for ever as it might prevent the thread to

   380                     // stop gracefully. millis will be 0 if no deadline was found.

   381                     int n = selector.select(millis == 0 ? NODEADLINE : millis);

   382                     if (n == 0) {

   383                         // Check whether client is still alive, and if not,

   384                         // gracefully stop this thread

   385                         if ((client = ownerRef.get()) == null) {

   386                             Log.logTrace("HttpClient no longer referenced. Exiting...");

   387                             return;

   388                         }

   389                         client.purgeTimeoutsAndReturnNextDeadline();

   390                         client = null; // don't hold onto the client ref

   391                         continue;

   392                     }

   393                     Set<SelectionKey> keys = selector.selectedKeys();

   394 

   395                     for (SelectionKey key : keys) {

   396                         SelectorAttachment sa = (SelectorAttachment) key.attachment();

   397                         int eventsOccurred = key.readyOps();

   398                         sa.events(eventsOccurred).forEach(readyList::add);

   399                         sa.resetInterestOps(eventsOccurred);

   400                     }

   401                     selector.selectNow(); // complete cancellation

   402                     selector.selectedKeys().clear();

   403 

   404                     for (AsyncEvent exchange : readyList) {

   405                         if (exchange.blocking()) {

   406                             exchange.channel().configureBlocking(true);

   407                         }

   408                         handleEvent(exchange); // will be delegated to executor

   409                     }

   410                     readyList.clear();

   411                 }

   412             } catch (Throwable e) {

   413                 if (!closed) {

   414                     // This terminates thread. So, better just print stack trace

   415                     String err = Utils.stackTrace(e);

   416                     Log.logError("HttpClientImpl: fatal error: " + err);

   417                 }

   418             } finally {

   419                 shutdown();

   420             }

   421         }

   422 

   423         void debugPrint(Selector selector) {

   424             System.err.println("Selector: debugprint start");

   425             Set<SelectionKey> keys = selector.keys();

   426             for (SelectionKey key : keys) {

   427                 SelectableChannel c = key.channel();

   428                 int ops = key.interestOps();

   429                 System.err.printf("selector chan:%s ops:%d\n", c, ops);

   430             }

   431             System.err.println("Selector: debugprint end");

   432         }

   433 

   434         void handleEvent(AsyncEvent e) {

   435             if (closed) {

   436                 e.abort();

   437             } else {

   438                 e.handle();

   439             }

   440         }

   441     }

   442 

   443     /**

   444      * Tracks multiple user level registrations associated with one NIO

   445      * registration (SelectionKey). In this implementation, registrations

   446      * are one-off and when an event is posted the registration is cancelled

   447      * until explicitly registered again.

   448      *

   449      * <p> No external synchronization required as this class is only used

   450      * by the SelectorManager thread. One of these objects required per

   451      * connection.

   452      */

   453     private static class SelectorAttachment {

   454         private final SelectableChannel chan;

   455         private final Selector selector;

   456         private final ArrayList<AsyncEvent> pending;

   457         private int interestOps;

   458 

   459         SelectorAttachment(SelectableChannel chan, Selector selector) {

   460             this.pending = new ArrayList<>();

   461             this.chan = chan;

   462             this.selector = selector;

   463         }

   464 

   465         void register(AsyncEvent e) throws ClosedChannelException {

   466             int newOps = e.interestOps();

   467             boolean reRegister = (interestOps & newOps) != newOps;

   468             interestOps |= newOps;

   469             pending.add(e);

   470             if (reRegister) {

   471                 // first time registration happens here also

   472                 chan.register(selector, interestOps, this);

   473             }

   474         }

   475 

   476         /**

   477          * Returns a Stream<AsyncEvents> containing only events that are

   478          * registered with the given {@code interestOps}.

   479          */

   480         Stream<AsyncEvent> events(int interestOps) {

   481             return pending.stream()

   482                     .filter(ev -> (ev.interestOps() & interestOps) != 0);

   483         }

   484 

   485         /**

   486          * Removes any events with the given {@code interestOps}, and if no

   487          * events remaining, cancels the associated SelectionKey.

   488          */

   489         void resetInterestOps(int interestOps) {

   490             int newOps = 0;

   491 

   492             Iterator<AsyncEvent> itr = pending.iterator();

   493             while (itr.hasNext()) {

   494                 AsyncEvent event = itr.next();

   495                 int evops = event.interestOps();

   496                 if (event.repeating()) {

   497                     newOps |= evops;

   498                     continue;

   499                 }

   500                 if ((evops & interestOps) != 0) {

   501                     itr.remove();

   502                 } else {

   503                     newOps |= evops;

   504                 }

   505             }

   506 

   507             this.interestOps = newOps;

   508             SelectionKey key = chan.keyFor(selector);

   509             if (newOps == 0) {

   510                 key.cancel();

   511             } else {

   512                 key.interestOps(newOps);

   513             }

   514         }

   515     }

   516 

   517     @Override

   518     public SSLContext sslContext() {

   519         Utils.checkNetPermission("getSSLContext");

   520         return sslContext;

   521     }

   522 

   523     @Override

   524     public Optional<SSLParameters> sslParameters() {

   525         return Optional.ofNullable(sslParams);

   526     }

   527 

   528     @Override

   529     public Optional<Authenticator> authenticator() {

   530         return Optional.ofNullable(authenticator);

   531     }

   532 

   533     @Override

   534     public Executor executor() {

   535         return executor;

   536     }

   537 

   538     ConnectionPool connectionPool() {

   539         return connections;

   540     }

   541 

   542     @Override

   543     public Redirect followRedirects() {

   544         return followRedirects;

   545     }

   546 

   547 

   548     @Override

   549     public Optional<CookieManager> cookieManager() {

   550         return Optional.ofNullable(cookieManager);

   551     }

   552 

   553     @Override

   554     public Optional<ProxySelector> proxy() {

   555         return Optional.ofNullable(this.proxySelector);

   556     }

   557 

   558     @Override

   559     public WebSocket.Builder newWebSocketBuilder(URI uri,

   560                                                  WebSocket.Listener listener) {

   561         return new BuilderImpl(this, uri, listener);

   562     }

   563 

   564     @Override

   565     public Version version() {

   566         return version;

   567     }

   568 

   569     //private final HashMap<String, Boolean> http2NotSupported = new HashMap<>();

   570 

   571     boolean getHttp2Allowed() {

   572         return version.equals(Version.HTTP_2);

   573     }

   574 

   575     private void initFilters() {

   576         addFilter(AuthenticationFilter.class);

   577         addFilter(RedirectFilter.class);

   578         if (this.cookieManager != null) {

   579             addFilter(CookieFilter.class);

   580         }

   581     }

   582 

   583     private void addFilter(Class<? extends HeaderFilter> f) {

   584         filters.addFilter(f);

   585     }

   586 

   587     final List<HeaderFilter> filterChain() {

   588         return filters.getFilterChain();

   589     }

   590 

   591     // Timer controls.

   592     // Timers are implemented through timed Selector.select() calls.

   593 

   594     synchronized void registerTimer(TimeoutEvent event) {

   595         Log.logTrace("Registering timer {0}", event);

   596         timeouts.add(event);

   597         selmgr.wakeupSelector();

   598     }

   599 

   600     synchronized void cancelTimer(TimeoutEvent event) {

   601         Log.logTrace("Canceling timer {0}", event);

   602         timeouts.remove(event);

   603     }

   604 

   605     /**

   606      * Purges ( handles ) timer events that have passed their deadline, and

   607      * returns the amount of time, in milliseconds, until the next earliest

   608      * event. A return value of 0 means that there are no events.

   609      */

   610     private long purgeTimeoutsAndReturnNextDeadline() {

   611         long diff = 0L;

   612         List<TimeoutEvent> toHandle = null;

   613         int remaining = 0;

   614         // enter critical section to retrieve the timeout event to handle

   615         synchronized(this) {

   616             if (timeouts.isEmpty()) return 0L;

   617 

   618             Instant now = Instant.now();

   619             Iterator<TimeoutEvent> itr = timeouts.iterator();

   620             while (itr.hasNext()) {

   621                 TimeoutEvent event = itr.next();

   622                 diff = now.until(event.deadline(), ChronoUnit.MILLIS);

   623                 if (diff <= 0) {

   624                     itr.remove();

   625                     toHandle = (toHandle == null) ? new ArrayList<>() : toHandle;

   626                     toHandle.add(event);

   627                 } else {

   628                     break;

   629                 }

   630             }

   631             remaining = timeouts.size();

   632         }

   633 

   634         // can be useful for debugging

   635         if (toHandle != null && Log.trace()) {

   636             Log.logTrace("purgeTimeoutsAndReturnNextDeadline: handling "

   637                     + (toHandle == null ? 0 : toHandle.size()) + " events, "

   638                     + "remaining " + remaining

   639                     + ", next deadline: " + (diff < 0 ? 0L : diff));

   640         }

   641 

   642         // handle timeout events out of critical section

   643         if (toHandle != null) {

   644             Throwable failed = null;

   645             for (TimeoutEvent event : toHandle) {

   646                 try {

   647                    Log.logTrace("Firing timer {0}", event);

   648                    event.handle();

   649                 } catch (Error | RuntimeException e) {

   650                     // Not expected. Handle remaining events then throw...

   651                     // If e is an OOME or SOE it might simply trigger a new

   652                     // error from here - but in this case there's not much we

   653                     // could do anyway. Just let it flow...

   654                     if (failed == null) failed = e;

   655                     else failed.addSuppressed(e);

   656                     Log.logTrace("Failed to handle event {0}: {1}", event, e);

   657                 }

   658             }

   659             if (failed instanceof Error) throw (Error) failed;

   660             if (failed instanceof RuntimeException) throw (RuntimeException) failed;

   661         }

   662 

   663         // return time to wait until next event. 0L if there's no more events.

   664         return diff < 0 ? 0L : diff;

   665     }

   666 

   667     // used for the connection window

   668     int getReceiveBufferSize() {

   669         return Utils.getIntegerNetProperty(

   670                 "jdk.httpclient.connectionWindowSize", 256 * 1024

   671         );

   672     }

   673 }