rev 13664 : 8139965: Hang seen when using com.sun.jndi.ldap.search.replyQueueSize
Reviewed-by: dfuchs

   1 /*
   2  * Copyright (c) 1999, 2018, 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 com.sun.jndi.ldap;
  27 
  28 import java.io.BufferedInputStream;
  29 import java.io.BufferedOutputStream;
  30 import java.io.IOException;
  31 import java.io.InputStream;
  32 import java.io.InterruptedIOException;
  33 import java.io.OutputStream;
  34 import java.lang.reflect.Constructor;
  35 import java.lang.reflect.InvocationTargetException;
  36 import java.lang.reflect.Method;
  37 import java.net.Socket;
  38 import java.security.AccessController;
  39 import java.security.PrivilegedAction;
  40 import java.util.Arrays;
  41 
  42 import javax.naming.CommunicationException;
  43 import javax.naming.InterruptedNamingException;
  44 import javax.naming.NamingException;
  45 import javax.naming.ServiceUnavailableException;
  46 import javax.naming.ldap.Control;
  47 import javax.net.ssl.SSLParameters;
  48 import javax.net.ssl.SSLSocket;
  49 
  50 import sun.misc.IOUtils;
  51 
  52 /**
  53   * A thread that creates a connection to an LDAP server.
  54   * After the connection, the thread reads from the connection.
  55   * A caller can invoke methods on the instance to read LDAP responses
  56   * and to send LDAP requests.
  57   * <p>
  58   * There is a one-to-one correspondence between an LdapClient and
  59   * a Connection. Access to Connection and its methods is only via
  60   * LdapClient with two exceptions: SASL authentication and StartTLS.
  61   * SASL needs to access Connection's socket IO streams (in order to do encryption
  62   * of the security layer). StartTLS needs to do replace IO streams
  63   * and close the IO  streams on nonfatal close. The code for SASL
  64   * authentication can be treated as being the same as from LdapClient
  65   * because the SASL code is only ever called from LdapClient, from
  66   * inside LdapClient's synchronized authenticate() method. StartTLS is called
  67   * directly by the application but should only occur when the underlying
  68   * connection is quiet.
  69   * <p>
  70   * In terms of synchronization, worry about data structures
  71   * used by the Connection thread because that usage might contend
  72   * with calls by the main threads (i.e., those that call LdapClient).
  73   * Main threads need to worry about contention with each other.
  74   * Fields that Connection thread uses:
  75   *     inStream - synced access and update; initialized in constructor;
  76   *           referenced outside class unsync'ed (by LdapSasl) only
  77   *           when connection is quiet
  78   *     traceFile, traceTagIn, traceTagOut - no sync; debugging only
  79   *     parent - no sync; initialized in constructor; no updates
  80   *     pendingRequests - sync
  81   *     pauseLock - per-instance lock;
  82   *     paused - sync via pauseLock (pauseReader())
  83   * Members used by main threads (LdapClient):
  84   *     host, port - unsync; read-only access for StartTLS and debug messages
  85   *     setBound(), setV3() - no sync; called only by LdapClient.authenticate(),
  86   *             which is a sync method called only when connection is "quiet"
  87   *     getMsgId() - sync
  88   *     writeRequest(), removeRequest(),findRequest(), abandonOutstandingReqs() -
  89   *             access to shared pendingRequests is sync
  90   *     writeRequest(),  abandonRequest(), ldapUnbind() - access to outStream sync
  91   *     cleanup() - sync
  92   *     readReply() - access to sock sync
  93   *     unpauseReader() - (indirectly via writeRequest) sync on pauseLock
  94   * Members used by SASL auth (main thread):
  95   *     inStream, outStream - no sync; used to construct new stream; accessed
  96   *             only when conn is "quiet" and not shared
  97   *     replaceStreams() - sync method
  98   * Members used by StartTLS:
  99   *     inStream, outStream - no sync; used to record the existing streams;
 100   *             accessed only when conn is "quiet" and not shared
 101   *     replaceStreams() - sync method
 102   * <p>
 103   * Handles anonymous, simple, and SASL bind for v3; anonymous and simple
 104   * for v2.
 105   * %%% made public for access by LdapSasl %%%
 106   *
 107   * @author Vincent Ryan
 108   * @author Rosanna Lee
 109   * @author Jagane Sundar
 110   */
 111 public final class Connection implements Runnable {
 112 
 113     private static final boolean debug = false;
 114     private static final int dump = 0; // > 0 r, > 1 rw
 115 
 116 
 117     final private Thread worker;    // Initialized in constructor
 118 
 119     private boolean v3 = true;       // Set in setV3()
 120 
 121     final public String host;  // used by LdapClient for generating exception messages
 122                          // used by StartTlsResponse when creating an SSL socket
 123     final public int port;     // used by LdapClient for generating exception messages
 124                          // used by StartTlsResponse when creating an SSL socket
 125 
 126     private boolean bound = false;   // Set in setBound()
 127 
 128     // All three are initialized in constructor and read-only afterwards
 129     private OutputStream traceFile = null;
 130     private String traceTagIn = null;
 131     private String traceTagOut = null;
 132 
 133     // Initialized in constructor; read and used externally (LdapSasl);
 134     // Updated in replaceStreams() during "quiet", unshared, period
 135     public InputStream inStream;   // must be public; used by LdapSasl
 136 
 137     // Initialized in constructor; read and used externally (LdapSasl);
 138     // Updated in replaceOutputStream() during "quiet", unshared, period
 139     public OutputStream outStream; // must be public; used by LdapSasl
 140 
 141     // Initialized in constructor; read and used externally (TLS) to
 142     // get new IO streams; closed during cleanup
 143     public Socket sock;            // for TLS
 144 
 145     // For processing "disconnect" unsolicited notification
 146     // Initialized in constructor
 147     final private LdapClient parent;
 148 
 149     // Incremented and returned in sync getMsgId()
 150     private int outMsgId = 0;
 151 
 152     //
 153     // The list of ldapRequests pending on this binding
 154     //
 155     // Accessed only within sync methods
 156     private LdapRequest pendingRequests = null;
 157 
 158     volatile IOException closureReason = null;
 159     volatile boolean useable = true;  // is Connection still useable
 160 
 161     int readTimeout;
 162     int connectTimeout;
 163     private static final boolean IS_HOSTNAME_VERIFICATION_DISABLED
 164             = hostnameVerificationDisabledValue();
 165 
 166     private static boolean hostnameVerificationDisabledValue() {
 167         PrivilegedAction<String> act = () -> System.getProperty(
 168                 "com.sun.jndi.ldap.object.disableEndpointIdentification");
 169         String prop = AccessController.doPrivileged(act);
 170         if (prop == null) {
 171             return false;
 172         }
 173         return prop.isEmpty() ? true : Boolean.parseBoolean(prop);
 174     }
 175     // true means v3; false means v2
 176     // Called in LdapClient.authenticate() (which is synchronized)
 177     // when connection is "quiet" and not shared; no need to synchronize
 178     void setV3(boolean v) {
 179         v3 = v;
 180     }
 181 
 182     // A BIND request has been successfully made on this connection
 183     // When cleaning up, remember to do an UNBIND
 184     // Called in LdapClient.authenticate() (which is synchronized)
 185     // when connection is "quiet" and not shared; no need to synchronize
 186     void setBound() {
 187         bound = true;
 188     }
 189 
 190     ////////////////////////////////////////////////////////////////////////////
 191     //
 192     // Create an LDAP Binding object and bind to a particular server
 193     //
 194     ////////////////////////////////////////////////////////////////////////////
 195 
 196     Connection(LdapClient parent, String host, int port, String socketFactory,
 197         int connectTimeout, int readTimeout, OutputStream trace) throws NamingException {
 198 
 199         this.host = host;
 200         this.port = port;
 201         this.parent = parent;
 202         this.readTimeout = readTimeout;
 203         this.connectTimeout = connectTimeout;
 204 
 205         if (trace != null) {
 206             traceFile = trace;
 207             traceTagIn = "<- " + host + ":" + port + "\n\n";
 208             traceTagOut = "-> " + host + ":" + port + "\n\n";
 209         }
 210 
 211         //
 212         // Connect to server
 213         //
 214         try {
 215             sock = createSocket(host, port, socketFactory, connectTimeout);
 216 
 217             if (debug) {
 218                 System.err.println("Connection: opening socket: " + host + "," + port);
 219             }
 220 
 221             inStream = new BufferedInputStream(sock.getInputStream());
 222             outStream = new BufferedOutputStream(sock.getOutputStream());
 223 
 224         } catch (InvocationTargetException e) {
 225             Throwable realException = e.getTargetException();
 226             // realException.printStackTrace();
 227 
 228             CommunicationException ce =
 229                 new CommunicationException(host + ":" + port);
 230             ce.setRootCause(realException);
 231             throw ce;
 232         } catch (Exception e) {
 233             // Class.forName() seems to do more error checking
 234             // and will throw IllegalArgumentException and such.
 235             // That's why we need to have a catch all here and
 236             // ignore generic exceptions.
 237             // Also catches all IO errors generated by socket creation.
 238             CommunicationException ce =
 239                 new CommunicationException(host + ":" + port);
 240             ce.setRootCause(e);
 241             throw ce;
 242         }
 243 
 244         worker = Obj.helper.createThread(this);
 245         worker.setDaemon(true);
 246         worker.start();
 247     }
 248 
 249     /*
 250      * Create an InetSocketAddress using the specified hostname and port number.
 251      */
 252     private Object createInetSocketAddress(String host, int port)
 253             throws NoSuchMethodException {
 254 
 255         try {
 256             Class<?> inetSocketAddressClass =
 257                 Class.forName("java.net.InetSocketAddress");
 258 
 259             Constructor<?> inetSocketAddressCons =
 260                 inetSocketAddressClass.getConstructor(new Class<?>[]{
 261                 String.class, int.class});
 262 
 263             return inetSocketAddressCons.newInstance(new Object[]{
 264                 host, new Integer(port)});
 265 
 266         } catch (ClassNotFoundException |
 267                  InstantiationException |
 268                  InvocationTargetException |
 269                  IllegalAccessException e) {
 270             throw new NoSuchMethodException();
 271 
 272         }
 273     }
 274 
 275     /*
 276      * Create a Socket object using the specified socket factory and time limit.
 277      *
 278      * If a timeout is supplied and unconnected sockets are supported then
 279      * an unconnected socket is created and the timeout is applied when
 280      * connecting the socket. If a timeout is supplied but unconnected sockets
 281      * are not supported then the timeout is ignored and a connected socket
 282      * is created.
 283      */
 284     private Socket createSocket(String host, int port, String socketFactory,
 285             int connectTimeout) throws Exception {
 286 
 287         Socket socket = null;
 288 
 289         if (socketFactory != null) {
 290 
 291             // create the factory
 292 
 293             Class<?> socketFactoryClass = Obj.helper.loadClass(socketFactory);
 294             Method getDefault =
 295                 socketFactoryClass.getMethod("getDefault", new Class<?>[]{});
 296             Object factory = getDefault.invoke(null, new Object[]{});
 297 
 298             // create the socket
 299 
 300             Method createSocket = null;
 301 
 302             if (connectTimeout > 0) {
 303 
 304                 try {
 305                     createSocket = socketFactoryClass.getMethod("createSocket",
 306                         new Class<?>[]{});
 307 
 308                     Method connect = Socket.class.getMethod("connect",
 309                         new Class<?>[]{Class.forName("java.net.SocketAddress"),
 310                         int.class});
 311                     Object endpoint = createInetSocketAddress(host, port);
 312 
 313                     // unconnected socket
 314                     socket =
 315                         (Socket)createSocket.invoke(factory, new Object[]{});
 316 
 317                     if (debug) {
 318                         System.err.println("Connection: creating socket with " +
 319                             "a timeout using supplied socket factory");
 320                     }
 321 
 322                     // connected socket
 323                     connect.invoke(socket, new Object[]{
 324                         endpoint, new Integer(connectTimeout)});
 325 
 326                 } catch (NoSuchMethodException e) {
 327                     // continue (but ignore connectTimeout)
 328                 }
 329             }
 330 
 331             if (socket == null) {
 332                 createSocket = socketFactoryClass.getMethod("createSocket",
 333                     new Class<?>[]{String.class, int.class});
 334 
 335                 if (debug) {
 336                     System.err.println("Connection: creating socket using " +
 337                         "supplied socket factory");
 338                 }
 339                 // connected socket
 340                 socket = (Socket) createSocket.invoke(factory,
 341                     new Object[]{host, new Integer(port)});
 342             }
 343         } else {
 344 
 345             if (connectTimeout > 0) {
 346 
 347                 try {
 348                     Constructor<Socket> socketCons =
 349                         Socket.class.getConstructor(new Class<?>[]{});
 350 
 351                     Method connect = Socket.class.getMethod("connect",
 352                         new Class<?>[]{Class.forName("java.net.SocketAddress"),
 353                         int.class});
 354                     Object endpoint = createInetSocketAddress(host, port);
 355 
 356                     socket = socketCons.newInstance(new Object[]{});
 357 
 358                     if (debug) {
 359                         System.err.println("Connection: creating socket with " +
 360                             "a timeout");
 361                     }
 362                     connect.invoke(socket, new Object[]{
 363                         endpoint, new Integer(connectTimeout)});
 364 
 365                 } catch (NoSuchMethodException e) {
 366                     // continue (but ignore connectTimeout)
 367                 }
 368             }
 369 
 370             if (socket == null) {
 371                 if (debug) {
 372                     System.err.println("Connection: creating socket");
 373                 }
 374                 // connected socket
 375                 socket = new Socket(host, port);
 376             }
 377         }
 378 
 379         // For LDAP connect timeouts on LDAP over SSL connections must treat
 380         // the SSL handshake following socket connection as part of the timeout.
 381         // So explicitly set a socket read timeout, trigger the SSL handshake,
 382         // then reset the timeout.
 383         if (socket instanceof SSLSocket) {
 384             SSLSocket sslSocket = (SSLSocket) socket;
 385             if (!IS_HOSTNAME_VERIFICATION_DISABLED) {
 386                 SSLParameters param = sslSocket.getSSLParameters();
 387                 param.setEndpointIdentificationAlgorithm("LDAPS");
 388                 sslSocket.setSSLParameters(param);
 389             }
 390             if (connectTimeout > 0) {
 391                 int socketTimeout = sslSocket.getSoTimeout();
 392                 sslSocket.setSoTimeout(connectTimeout); // reuse full timeout value
 393                 sslSocket.startHandshake();
 394                 sslSocket.setSoTimeout(socketTimeout);
 395             }
 396         }
 397         return socket;
 398     }
 399 
 400     ////////////////////////////////////////////////////////////////////////////
 401     //
 402     // Methods to IO to the LDAP server
 403     //
 404     ////////////////////////////////////////////////////////////////////////////
 405 
 406     synchronized int getMsgId() {
 407         return ++outMsgId;
 408     }
 409 
 410     LdapRequest writeRequest(BerEncoder ber, int msgId) throws IOException {
 411         return writeRequest(ber, msgId, false /* pauseAfterReceipt */, -1);
 412     }
 413 
 414     LdapRequest writeRequest(BerEncoder ber, int msgId,
 415         boolean pauseAfterReceipt) throws IOException {
 416         return writeRequest(ber, msgId, pauseAfterReceipt, -1);
 417     }
 418 
 419     LdapRequest writeRequest(BerEncoder ber, int msgId,
 420         boolean pauseAfterReceipt, int replyQueueCapacity) throws IOException {
 421 
 422         LdapRequest req =
 423             new LdapRequest(msgId, pauseAfterReceipt, replyQueueCapacity);
 424         addRequest(req);
 425 
 426         if (traceFile != null) {
 427             Ber.dumpBER(traceFile, traceTagOut, ber.getBuf(), 0, ber.getDataLen());
 428         }
 429 
 430 
 431         // unpause reader so that it can get response
 432         // NOTE: Must do this before writing request, otherwise might
 433         // create a race condition where the writer unblocks its own response
 434         unpauseReader();
 435 
 436         if (debug) {
 437             System.err.println("Writing request to: " + outStream);
 438         }
 439 
 440         try {
 441             synchronized (this) {
 442                 outStream.write(ber.getBuf(), 0, ber.getDataLen());
 443                 outStream.flush();
 444             }
 445         } catch (IOException e) {
 446             cleanup(null, true);
 447             throw (closureReason = e); // rethrow
 448         }
 449 
 450         return req;
 451     }
 452 
 453     /**
 454      * Reads a reply; waits until one is ready.
 455      */
 456     BerDecoder readReply(LdapRequest ldr)
 457             throws IOException, NamingException {
 458         BerDecoder rber;
 459 
 460         // Track down elapsed time to workaround spurious wakeups
 461         long elapsedMilli = 0;
 462         long elapsedNano = 0;
 463 
 464         while (((rber = ldr.getReplyBer()) == null) &&
 465                 (readTimeout <= 0 || elapsedMilli < readTimeout))
 466         {
 467             try {
 468                 // If socket closed, don't even try
 469                 synchronized (this) {
 470                     if (sock == null) {
 471                         throw new ServiceUnavailableException(host + ":" + port +
 472                             "; socket closed");
 473                     }
 474                 }
 475                 synchronized (ldr) {
 476                     // check if condition has changed since our last check
 477                     rber = ldr.getReplyBer();
 478                     if (rber == null) {
 479                         if (readTimeout > 0) {  // Socket read timeout is specified
 480                             long beginNano = System.nanoTime();
 481 
 482                             // will be woken up before readTimeout if reply is
 483                             // available
 484                             ldr.wait(readTimeout - elapsedMilli);
 485                             elapsedNano += (System.nanoTime() - beginNano);
 486                             elapsedMilli += elapsedNano / 1000_000;
 487                             elapsedNano %= 1000_000;
 488 
 489                         } else {
 490                             // no timeout is set so we wait infinitely until
 491                             // a response is received
 492                             // https://docs.oracle.com/javase/8/docs/technotes/guides/jndi/jndi-ldap.html#PROP
 493                             ldr.wait();
 494                         }
 495                     } else {
 496                         break;
 497                     }
 498                 }
 499             } catch (InterruptedException ex) {
 500                 throw new InterruptedNamingException(
 501                     "Interrupted during LDAP operation");
 502             }
 503         }
 504 
 505         if ((rber == null) && (elapsedMilli >= readTimeout)) {
 506             abandonRequest(ldr, null);
 507             throw new NamingException("LDAP response read timed out, timeout used:"

 508                             + readTimeout + "ms." );
 509 
 510         }
 511         return rber;
 512     }
 513 
 514 
 515     ////////////////////////////////////////////////////////////////////////////
 516     //
 517     // Methods to add, find, delete, and abandon requests made to server
 518     //
 519     ////////////////////////////////////////////////////////////////////////////
 520 
 521     private synchronized void addRequest(LdapRequest ldapRequest) {
 522 
 523         LdapRequest ldr = pendingRequests;
 524         if (ldr == null) {
 525             pendingRequests = ldapRequest;
 526             ldapRequest.next = null;
 527         } else {
 528             ldapRequest.next = pendingRequests;
 529             pendingRequests = ldapRequest;
 530         }
 531     }
 532 
 533     synchronized LdapRequest findRequest(int msgId) {
 534 
 535         LdapRequest ldr = pendingRequests;
 536         while (ldr != null) {
 537             if (ldr.msgId == msgId) {
 538                 return ldr;
 539             }
 540             ldr = ldr.next;
 541         }
 542         return null;
 543 
 544     }
 545 
 546     synchronized void removeRequest(LdapRequest req) {
 547         LdapRequest ldr = pendingRequests;
 548         LdapRequest ldrprev = null;
 549 
 550         while (ldr != null) {
 551             if (ldr == req) {
 552                 ldr.cancel();
 553 
 554                 if (ldrprev != null) {
 555                     ldrprev.next = ldr.next;
 556                 } else {
 557                     pendingRequests = ldr.next;
 558                 }
 559                 ldr.next = null;
 560             }
 561             ldrprev = ldr;
 562             ldr = ldr.next;
 563         }
 564     }
 565 
 566     void abandonRequest(LdapRequest ldr, Control[] reqCtls) {
 567         // Remove from queue
 568         removeRequest(ldr);
 569 
 570         BerEncoder ber = new BerEncoder(256);
 571         int abandonMsgId = getMsgId();
 572 
 573         //
 574         // build the abandon request.
 575         //
 576         try {
 577             ber.beginSeq(Ber.ASN_SEQUENCE | Ber.ASN_CONSTRUCTOR);
 578                 ber.encodeInt(abandonMsgId);
 579                 ber.encodeInt(ldr.msgId, LdapClient.LDAP_REQ_ABANDON);
 580 
 581                 if (v3) {
 582                     LdapClient.encodeControls(ber, reqCtls);
 583                 }
 584             ber.endSeq();
 585 
 586             if (traceFile != null) {
 587                 Ber.dumpBER(traceFile, traceTagOut, ber.getBuf(), 0,
 588                     ber.getDataLen());
 589             }
 590 
 591             synchronized (this) {
 592                 outStream.write(ber.getBuf(), 0, ber.getDataLen());
 593                 outStream.flush();
 594             }
 595 
 596         } catch (IOException ex) {
 597             //System.err.println("ldap.abandon: " + ex);
 598         }
 599 
 600         // Don't expect any response for the abandon request.
 601     }
 602 
 603     synchronized void abandonOutstandingReqs(Control[] reqCtls) {
 604         LdapRequest ldr = pendingRequests;
 605 
 606         while (ldr != null) {
 607             abandonRequest(ldr, reqCtls);
 608             pendingRequests = ldr = ldr.next;
 609         }
 610     }
 611 
 612     ////////////////////////////////////////////////////////////////////////////
 613     //
 614     // Methods to unbind from server and clear up resources when object is
 615     // destroyed.
 616     //
 617     ////////////////////////////////////////////////////////////////////////////
 618 
 619     private void ldapUnbind(Control[] reqCtls) {
 620 
 621         BerEncoder ber = new BerEncoder(256);
 622         int unbindMsgId = getMsgId();
 623 
 624         //
 625         // build the unbind request.
 626         //
 627 
 628         try {
 629 
 630             ber.beginSeq(Ber.ASN_SEQUENCE | Ber.ASN_CONSTRUCTOR);
 631                 ber.encodeInt(unbindMsgId);
 632                 // IMPLICIT TAGS
 633                 ber.encodeByte(LdapClient.LDAP_REQ_UNBIND);
 634                 ber.encodeByte(0);
 635 
 636                 if (v3) {
 637                     LdapClient.encodeControls(ber, reqCtls);
 638                 }
 639             ber.endSeq();
 640 
 641             if (traceFile != null) {
 642                 Ber.dumpBER(traceFile, traceTagOut, ber.getBuf(),
 643                     0, ber.getDataLen());
 644             }
 645 
 646             synchronized (this) {
 647                 outStream.write(ber.getBuf(), 0, ber.getDataLen());
 648                 outStream.flush();
 649             }
 650 
 651         } catch (IOException ex) {
 652             //System.err.println("ldap.unbind: " + ex);
 653         }
 654 
 655         // Don't expect any response for the unbind request.
 656     }
 657 
 658     /**
 659      * @param reqCtls Possibly null request controls that accompanies the
 660      *    abandon and unbind LDAP request.
 661      * @param notifyParent true means to call parent LdapClient back, notifying
 662      *    it that the connection has been closed; false means not to notify
 663      *    parent. If LdapClient invokes cleanup(), notifyParent should be set to
 664      *    false because LdapClient already knows that it is closing
 665      *    the connection. If Connection invokes cleanup(), notifyParent should be
 666      *    set to true because LdapClient needs to know about the closure.
 667      */
 668     void cleanup(Control[] reqCtls, boolean notifyParent) {
 669         boolean nparent = false;
 670 
 671         synchronized (this) {
 672             useable = false;
 673 
 674             if (sock != null) {
 675                 if (debug) {
 676                     System.err.println("Connection: closing socket: " + host + "," + port);
 677                 }
 678                 try {
 679                     if (!notifyParent) {
 680                         abandonOutstandingReqs(reqCtls);
 681                     }
 682                     if (bound) {
 683                         ldapUnbind(reqCtls);
 684                     }
 685                 } finally {
 686                     try {
 687                         outStream.flush();
 688                         sock.close();
 689                         unpauseReader();
 690                     } catch (IOException ie) {
 691                         if (debug)
 692                             System.err.println("Connection: problem closing socket: " + ie);
 693                     }
 694                     if (!notifyParent) {
 695                         LdapRequest ldr = pendingRequests;
 696                         while (ldr != null) {
 697                             ldr.cancel();
 698                             ldr = ldr.next;
 699                         }
 700                     }
 701                     sock = null;
 702                 }
 703                 nparent = notifyParent;
 704             }
 705             if (nparent) {
 706                 LdapRequest ldr = pendingRequests;
 707                 while (ldr != null) {
 708 
 709                     synchronized (ldr) {
 710                         ldr.notify();
 711                         ldr = ldr.next;
 712                     }
 713                 }
 714             }
 715         }
 716         if (nparent) {
 717             parent.processConnectionClosure();
 718         }
 719     }
 720 
 721 
 722     // Assume everything is "quiet"
 723     // "synchronize" might lead to deadlock so don't synchronize method
 724     // Use streamLock instead for synchronizing update to stream
 725 
 726     synchronized public void replaceStreams(InputStream newIn, OutputStream newOut) {
 727         if (debug) {
 728             System.err.println("Replacing " + inStream + " with: " + newIn);
 729             System.err.println("Replacing " + outStream + " with: " + newOut);
 730         }
 731 
 732         inStream = newIn;
 733 
 734         // Cleanup old stream
 735         try {
 736             outStream.flush();
 737         } catch (IOException ie) {
 738             if (debug)
 739                 System.err.println("Connection: cannot flush outstream: " + ie);
 740         }
 741 
 742         // Replace stream
 743         outStream = newOut;
 744     }
 745 
 746     /**
 747      * Used by Connection thread to read inStream into a local variable.
 748      * This ensures that there is no contention between the main thread
 749      * and the Connection thread when the main thread updates inStream.
 750      */
 751     synchronized private InputStream getInputStream() {
 752         return inStream;
 753     }
 754 
 755 
 756     ////////////////////////////////////////////////////////////////////////////
 757     //
 758     // Code for pausing/unpausing the reader thread ('worker')
 759     //
 760     ////////////////////////////////////////////////////////////////////////////
 761 
 762     /*
 763      * The main idea is to mark requests that need the reader thread to
 764      * pause after getting the response. When the reader thread gets the response,
 765      * it waits on a lock instead of returning to the read(). The next time a
 766      * request is sent, the reader is automatically unblocked if necessary.
 767      * Note that the reader must be unblocked BEFORE the request is sent.
 768      * Otherwise, there is a race condition where the request is sent and
 769      * the reader thread might read the response and be unblocked
 770      * by writeRequest().
 771      *
 772      * This pause gives the main thread (StartTLS or SASL) an opportunity to
 773      * update the reader's state (e.g., its streams) if necessary.
 774      * The assumption is that the connection will remain quiet during this pause
 775      * (i.e., no intervening requests being sent).
 776      *<p>
 777      * For dealing with StartTLS close,
 778      * when the read() exits either due to EOF or an exception,
 779      * the reader thread checks whether there is a new stream to read from.
 780      * If so, then it reattempts the read. Otherwise, the EOF or exception
 781      * is processed and the reader thread terminates.
 782      * In a StartTLS close, the client first replaces the SSL IO streams with
 783      * plain ones and then closes the SSL socket.
 784      * If the reader thread attempts to read, or was reading, from
 785      * the SSL socket (that is, it got to the read BEFORE replaceStreams()),
 786      * the SSL socket close will cause the reader thread to
 787      * get an EOF/exception and reexamine the input stream.
 788      * If the reader thread sees a new stream, it reattempts the read.
 789      * If the underlying socket is still alive, then the new read will succeed.
 790      * If the underlying socket has been closed also, then the new read will
 791      * fail and the reader thread exits.
 792      * If the reader thread attempts to read, or was reading, from the plain
 793      * socket (that is, it got to the read AFTER replaceStreams()), the
 794      * SSL socket close will have no effect on the reader thread.
 795      *
 796      * The check for new stream is made only
 797      * in the first attempt at reading a BER buffer; the reader should
 798      * never be in midst of reading a buffer when a nonfatal close occurs.
 799      * If this occurs, then the connection is in an inconsistent state and
 800      * the safest thing to do is to shut it down.
 801      */
 802 
 803     private Object pauseLock = new Object();  // lock for reader to wait on while paused
 804     private boolean paused = false;           // paused state of reader
 805 
 806     /*
 807      * Unpauses reader thread if it was paused
 808      */
 809     private void unpauseReader() throws IOException {
 810         synchronized (pauseLock) {
 811             if (paused) {
 812                 if (debug) {
 813                     System.err.println("Unpausing reader; read from: " +
 814                                         inStream);
 815                 }
 816                 paused = false;
 817                 pauseLock.notify();
 818             }
 819         }
 820     }
 821 
 822      /*
 823      * Pauses reader so that it stops reading from the input stream.
 824      * Reader blocks on pauseLock instead of read().
 825      * MUST be called from within synchronized (pauseLock) clause.
 826      */
 827     private void pauseReader() throws IOException {
 828         if (debug) {
 829             System.err.println("Pausing reader;  was reading from: " +
 830                                 inStream);
 831         }
 832         paused = true;
 833         try {
 834             while (paused) {
 835                 pauseLock.wait(); // notified by unpauseReader
 836             }
 837         } catch (InterruptedException e) {
 838             throw new InterruptedIOException(
 839                     "Pause/unpause reader has problems.");
 840         }
 841     }
 842 
 843 
 844     ////////////////////////////////////////////////////////////////////////////
 845     //
 846     // The LDAP Binding thread. It does the mux/demux of multiple requests
 847     // on the same TCP connection.
 848     //
 849     ////////////////////////////////////////////////////////////////////////////
 850 
 851 
 852     public void run() {
 853         byte inbuf[];   // Buffer for reading incoming bytes
 854         int inMsgId;    // Message id of incoming response
 855         int bytesread;  // Number of bytes in inbuf
 856         int br;         // Temp; number of bytes read from stream
 857         int offset;     // Offset of where to store bytes in inbuf
 858         int seqlen;     // Length of ASN sequence
 859         int seqlenlen;  // Number of sequence length bytes
 860         boolean eos;    // End of stream
 861         BerDecoder retBer;    // Decoder for ASN.1 BER data from inbuf
 862         InputStream in = null;
 863 
 864         try {
 865             while (true) {
 866                 try {
 867                     // type and length (at most 128 octets for long form)
 868                     inbuf = new byte[129];
 869 
 870                     offset = 0;
 871                     seqlen = 0;
 872                     seqlenlen = 0;
 873 
 874                     in = getInputStream();
 875 
 876                     // check that it is the beginning of a sequence
 877                     bytesread = in.read(inbuf, offset, 1);
 878                     if (bytesread < 0) {
 879                         if (in != getInputStream()) {
 880                             continue;   // a new stream to try
 881                         } else {
 882                             break; // EOF
 883                         }
 884                     }
 885 
 886                     if (inbuf[offset++] != (Ber.ASN_SEQUENCE | Ber.ASN_CONSTRUCTOR))
 887                         continue;
 888 
 889                     // get length of sequence
 890                     bytesread = in.read(inbuf, offset, 1);
 891                     if (bytesread < 0)
 892                         break; // EOF
 893                     seqlen = inbuf[offset++];
 894 
 895                     // if high bit is on, length is encoded in the
 896                     // subsequent length bytes and the number of length bytes
 897                     // is equal to & 0x80 (i.e. length byte with high bit off).
 898                     if ((seqlen & 0x80) == 0x80) {
 899                         seqlenlen = seqlen & 0x7f;  // number of length bytes
 900 
 901                         bytesread = 0;
 902                         eos = false;
 903 
 904                         // Read all length bytes
 905                         while (bytesread < seqlenlen) {
 906                             br = in.read(inbuf, offset+bytesread,
 907                                 seqlenlen-bytesread);
 908                             if (br < 0) {
 909                                 eos = true;
 910                                 break; // EOF
 911                             }
 912                             bytesread += br;
 913                         }
 914 
 915                         // end-of-stream reached before length bytes are read
 916                         if (eos)
 917                             break;  // EOF
 918 
 919                         // Add contents of length bytes to determine length
 920                         seqlen = 0;
 921                         for( int i = 0; i < seqlenlen; i++) {
 922                             seqlen = (seqlen << 8) + (inbuf[offset+i] & 0xff);
 923                         }
 924                         offset += bytesread;
 925                     }
 926 
 927                     // read in seqlen bytes
 928                     byte[] left = IOUtils.readFully(in, seqlen, false);
 929                     inbuf = Arrays.copyOf(inbuf, offset + left.length);
 930                     System.arraycopy(left, 0, inbuf, offset, left.length);
 931                     offset += left.length;
 932 /*
 933 if (dump > 0) {
 934 System.err.println("seqlen: " + seqlen);
 935 System.err.println("bufsize: " + offset);
 936 System.err.println("bytesleft: " + bytesleft);
 937 System.err.println("bytesread: " + bytesread);
 938 }
 939 */
 940 
 941 
 942                     try {
 943                         retBer = new BerDecoder(inbuf, 0, offset);
 944 
 945                         if (traceFile != null) {
 946                             Ber.dumpBER(traceFile, traceTagIn, inbuf, 0, offset);
 947                         }
 948 
 949                         retBer.parseSeq(null);
 950                         inMsgId = retBer.parseInt();
 951                         retBer.reset(); // reset offset
 952 
 953                         boolean needPause = false;
 954 
 955                         if (inMsgId == 0) {
 956                             // Unsolicited Notification
 957                             parent.processUnsolicited(retBer);
 958                         } else {
 959                             LdapRequest ldr = findRequest(inMsgId);
 960 
 961                             if (ldr != null) {
 962 
 963                                 /**
 964                                  * Grab pauseLock before making reply available
 965                                  * to ensure that reader goes into paused state
 966                                  * before writer can attempt to unpause reader
 967                                  */
 968                                 synchronized (pauseLock) {
 969                                     needPause = ldr.addReplyBer(retBer);
 970                                     if (needPause) {
 971                                         /*
 972                                          * Go into paused state; release
 973                                          * pauseLock
 974                                          */
 975                                         pauseReader();
 976                                     }
 977 
 978                                     // else release pauseLock
 979                                 }
 980                             } else {
 981                                 // System.err.println("Cannot find" +
 982                                 //              "LdapRequest for " + inMsgId);
 983                             }
 984                         }
 985                     } catch (Ber.DecodeException e) {
 986                         //System.err.println("Cannot parse Ber");
 987                     }
 988                 } catch (IOException ie) {
 989                     if (debug) {
 990                         System.err.println("Connection: Inside Caught " + ie);
 991                         ie.printStackTrace();
 992                     }
 993 
 994                     if (in != getInputStream()) {
 995                         // A new stream to try
 996                         // Go to top of loop and continue
 997                     } else {
 998                         if (debug) {
 999                             System.err.println("Connection: rethrowing " + ie);
1000                         }
1001                         throw ie;  // rethrow exception
1002                     }
1003                 }
1004             }
1005 
1006             if (debug) {
1007                 System.err.println("Connection: end-of-stream detected: "
1008                     + in);
1009             }
1010         } catch (IOException ex) {
1011             if (debug) {
1012                 System.err.println("Connection: Caught " + ex);
1013             }
1014             closureReason = ex;
1015         } finally {
1016             cleanup(null, true); // cleanup
1017         }
1018         if (debug) {
1019             System.err.println("Connection: Thread Exiting");
1020         }
1021     }
1022 
1023 
1024     // This code must be uncommented to run the LdapAbandonTest.
1025     /*public void sendSearchReqs(String dn, int numReqs) {
1026         int i;
1027         String attrs[] = null;
1028         for(i = 1; i <= numReqs; i++) {
1029             BerEncoder ber = new BerEncoder(2048);
1030 
1031             try {
1032             ber.beginSeq(Ber.ASN_SEQUENCE | Ber.ASN_CONSTRUCTOR);
1033                 ber.encodeInt(i);
1034                 ber.beginSeq(LdapClient.LDAP_REQ_SEARCH);
1035                     ber.encodeString(dn == null ? "" : dn);
1036                     ber.encodeInt(0, LdapClient.LBER_ENUMERATED);
1037                     ber.encodeInt(3, LdapClient.LBER_ENUMERATED);
1038                     ber.encodeInt(0);
1039                     ber.encodeInt(0);
1040                     ber.encodeBoolean(true);
1041                     LdapClient.encodeFilter(ber, "");
1042                     ber.beginSeq(Ber.ASN_SEQUENCE | Ber.ASN_CONSTRUCTOR);
1043                         ber.encodeStringArray(attrs);
1044                     ber.endSeq();
1045                 ber.endSeq();
1046             ber.endSeq();
1047             writeRequest(ber, i);
1048             //System.err.println("wrote request " + i);
1049             } catch (Exception ex) {
1050             //System.err.println("ldap.search: Caught " + ex + " building req");
1051             }
1052 
1053         }
1054     } */
1055 }
--- EOF ---