Fix OPENDJ-520: Worker threads are too greedy when caching memory used for ...

Matthew Swift

13.05.2012 9ff69f265dae8f0647fb9c204fda070eafe25613

Fix OPENDJ-520: Worker threads are too greedy when caching memory used for encoding/decoding entries and protocol messages

* introduce global option max-internal-buffer-size (default 32KB) for managing the maximum size of internal buffers
* use simpler ASN1Writer implementation for LDAP client connections
* minor fix to SASL confidentiality and integrity for large packets.

 opends/resource/schema/02-config.ldif

@@ -3326,6 +3326,12 @@
  NO-USER-MODIFICATION
  USAGE directoryOperation
  X-ORIGIN 'OpenDJ Directory Server' )
attributeTypes: ( 1.3.6.1.4.1.36733.2.1.1.61
  NAME 'ds-cfg-max-internal-buffer-size'
  EQUALITY caseIgnoreMatch
  SYNTAX 1.3.6.1.4.1.1466.115.121.1.15
  SINGLE-VALUE
  X-ORIGIN 'OpenDJ Directory Server' )
objectClasses: ( 1.3.6.1.4.1.26027.1.2.1
  NAME 'ds-cfg-access-control-handler'
  SUP top
@@ -3737,7 +3743,8 @@
        ds-cfg-etime-resolution $
        ds-cfg-entry-cache-preload $
        ds-cfg-max-allowed-client-connections $
        ds-cfg-max-psearches )
        ds-cfg-max-psearches $
        ds-cfg-max-internal-buffer-size )
  X-ORIGIN 'OpenDS Directory Server' )
objectClasses: ( 1.3.6.1.4.1.26027.1.2.40
  NAME 'ds-cfg-root-dn-user'

 opends/src/admin/defn/org/opends/server/admin/std/GlobalConfiguration.xml

@@ -24,7 +24,7 @@
  !
  !
  !      Copyright 2007-2010 Sun Microsystems, Inc.
  !      Portions Copyright 2011 ForgeRock AS
  !      Portions Copyright 2011-2012 ForgeRock AS
  ! -->
<adm:managed-object name="global" plural-name="globals"
  package="org.opends.server.admin.std"
@@ -895,4 +895,32 @@
      </ldap:attribute>
    </adm:profile>
  </adm:property>
  <adm:property name="max-internal-buffer-size" advanced="true">
    <adm:synopsis>
      The threshold capacity beyond which internal cached buffers used for
      encoding and decoding entries and protocol messages will be trimmed
      after use.
    </adm:synopsis>
    <adm:description>
      Individual buffers may grow very large when encoding and decoding
      large entries and protocol messages and should be reduced in size when
      they are no longer needed. This setting specifies the threshold at which
      a buffer is determined to have grown too big and should be trimmed down
      after use.  
    </adm:description>
    <adm:default-behavior>
      <adm:defined>
        <adm:value>32 KB</adm:value>
      </adm:defined>
    </adm:default-behavior>
    <adm:syntax>
      <!--  Upper limit to force 32-bit value -->
      <adm:size lower-limit="512 B" upper-limit="1 GB"/>
    </adm:syntax>
    <adm:profile name="ldap">
      <ldap:attribute>
        <ldap:name>ds-cfg-max-internal-buffer-size</ldap:name>
      </ldap:attribute>
    </adm:profile>
  </adm:property>
</adm:managed-object>

 opends/src/admin/messages/GlobalCfgDefn.properties

@@ -53,6 +53,8 @@
property.lookthrough-limit.description=This includes any entry that the server must examine in the course of processing the request, regardless of whether it actually matches the search criteria. A value of 0 indicates that no lookthrough limit is enforced. Note that this is the default server-wide limit, but it may be overridden on a per-user basis using the ds-rlim-lookthrough-limit operational attribute.
property.max-allowed-client-connections.synopsis=Specifies the maximum number of client connections that may be established at any given time
property.max-allowed-client-connections.description=A value of 0 indicates that unlimited client connection is allowed.
property.max-internal-buffer-size.synopsis=The threshold capacity beyond which internal cached buffers used for encoding and decoding entries and protocol messages will be trimmed after use.
property.max-internal-buffer-size.description=Individual buffers may grow very large when encoding and decoding large entries and protocol messages and should be reduced in size when they are no longer needed. This setting specifies the threshold at which a buffer is determined to have grown too big and should be trimmed down after use.
property.max-psearches.synopsis=Defines the maximum number of concurrent persistent searches that can be performed on directory server
property.max-psearches.description=The persistent search mechanism provides an active channel through which entries that change, and information about the changes that occur, can be communicated. Because each persistent search operation consumes resources, limiting the number of simultaneous persistent searches keeps the performance impact minimal. A value of -1 indicates that there is no limit on the persistent searches.
property.notify-abandoned-operations.synopsis=Indicates whether the directory server should send a response to any operation that is interrupted via an abandon request.

 opends/src/server/org/opends/server/backends/jeb/ID2Entry.java

@@ -23,11 +23,14 @@
 *
 *
 *      Copyright 2006-2010 Sun Microsystems, Inc.
 *      Portions copyright 2012 ForgeRock AS.
 */
package org.opends.server.backends.jeb;
import org.opends.messages.Message;

import static org.opends.server.core.DirectoryServer.getMaxInternalBufferSize;
import static org.opends.server.loggers.debug.DebugLogger.*;

import org.opends.server.loggers.debug.DebugTracer;
import static org.opends.messages.JebMessages.*;

@@ -61,30 +64,83 @@
   */
  private DataConfig dataConfig;

  private static ThreadLocal<EntryCoder> entryCodingBuffers =
      new ThreadLocal<EntryCoder>();
  /**
   * Cached encoding buffers.
   */
  private static final ThreadLocal<EntryCodec> ENTRY_CODEC_CACHE =
      new ThreadLocal<EntryCodec>()
  {
    protected EntryCodec initialValue()
    {
      return new EntryCodec();
    };
  };

  private static EntryCodec acquireEntryCodec()
  {
    EntryCodec codec = ENTRY_CODEC_CACHE.get();
    if (codec.maxBufferSize != getMaxInternalBufferSize())
    {
      // Setting has changed, so recreate the codec.
      codec = new EntryCodec();
      ENTRY_CODEC_CACHE.set(codec);
    }
    return codec;
  }

  /**
   * A cached set of ByteStringBuilder buffers and ASN1Writer used to encode
   * entries.
   */
  private static class EntryCoder
  private static class EntryCodec
  {
    ByteStringBuilder encodedBuffer;
    private ByteStringBuilder entryBuffer;
    private ByteStringBuilder compressedEntryBuffer;
    private ASN1Writer writer;
    private static final int BUFFER_INIT_SIZE = 512;

    private EntryCoder()
    private final ByteStringBuilder encodedBuffer = new ByteStringBuilder();
    private final ByteStringBuilder entryBuffer = new ByteStringBuilder();
    private final ByteStringBuilder compressedEntryBuffer =
        new ByteStringBuilder();
    private final ASN1Writer writer;
    private final int maxBufferSize;

    private EntryCodec()
    {
      encodedBuffer = new ByteStringBuilder();
      entryBuffer = new ByteStringBuilder();
      compressedEntryBuffer = new ByteStringBuilder();
      writer = ASN1.getWriter(encodedBuffer);
      this.maxBufferSize = getMaxInternalBufferSize();
      this.writer = ASN1.getWriter(encodedBuffer, maxBufferSize);
    }

    private void release()
    {
      try
      {
        writer.close(); // Clears encodedBuffer as well.
      }
      catch (Exception ignored)
      {
        // Unreachable.
      }

      if (entryBuffer.capacity() < maxBufferSize)
      {
        entryBuffer.clear();
      }
      else
      {
        entryBuffer.clear(BUFFER_INIT_SIZE);
      }

      if (compressedEntryBuffer.capacity() < maxBufferSize)
      {
        compressedEntryBuffer.clear();
      }
      else
      {
        compressedEntryBuffer.clear(BUFFER_INIT_SIZE);
      }
    }

    private Entry decode(ByteString bytes, CompressedSchema compressedSchema)
        throws DirectoryException,ASN1Exception,LDAPException,
        throws DirectoryException, ASN1Exception, LDAPException,
        DataFormatException, IOException
    {
      // Get the format version.
@@ -102,14 +158,8 @@

      // See if it was compressed.
      int uncompressedSize = (int)reader.readInteger();

      if(uncompressedSize > 0)
      {
        // We will use the cached buffers to avoid allocations.
        // Reset the buffers;
        entryBuffer.clear();
        compressedEntryBuffer.clear();

        // It was compressed.
        reader.readOctetString(compressedEntryBuffer);
        CryptoManager cryptoManager = DirectoryServer.getCryptoManager();
@@ -125,7 +175,7 @@
      else
      {
        // Since we don't have to do any decompression, we can just decode
        // the entry off the
        // the entry directly.
        ByteString encodedEntry = reader.readOctetString();
        return Entry.decode(encodedEntry.asReader(), compressedSchema);
      }
@@ -149,11 +199,6 @@
    private void encodeVolatile(Entry entry, DataConfig dataConfig)
        throws DirectoryException
    {
      // Reset the buffers;
      encodedBuffer.clear();
      entryBuffer.clear();
      compressedEntryBuffer.clear();

      // Encode the entry for later use.
      entry.encode(entryBuffer, dataConfig.getEntryEncodeConfig());

@@ -189,7 +234,6 @@
        {
          TRACER.debugCaught(DebugLogLevel.ERROR, ioe);
        }

      }
    }
  }
@@ -270,18 +314,19 @@
   * @throws DirectoryException If a Directory Server error occurs.
   * @throws IOException if an error occurs while reading the ASN1 sequence.
   */
  static public Entry entryFromDatabase(ByteString bytes,
                                        CompressedSchema compressedSchema)
      throws DirectoryException,ASN1Exception,LDAPException,
      DataFormatException,IOException
  public static Entry entryFromDatabase(ByteString bytes,
      CompressedSchema compressedSchema) throws DirectoryException,
      ASN1Exception, LDAPException, DataFormatException, IOException
  {
    EntryCoder coder = entryCodingBuffers.get();
    if(coder == null)
    EntryCodec codec = acquireEntryCodec();
    try
    {
      coder = new EntryCoder();
      entryCodingBuffers.set(coder);
      return codec.decode(bytes, compressedSchema);
    }
    return coder.decode(bytes, compressedSchema);
    finally
    {
      codec.release();
    }
  }

  /**
@@ -294,19 +339,22 @@
   * @throws  DirectoryException  If a problem occurs while attempting to encode
   *                              the entry.
   */
  static public ByteString entryToDatabase(Entry entry, DataConfig dataConfig)
  public static ByteString entryToDatabase(Entry entry, DataConfig dataConfig)
      throws DirectoryException
  {
    EntryCoder coder = entryCodingBuffers.get();
    if(coder == null)
    EntryCodec codec = acquireEntryCodec();
    try
    {
      coder = new EntryCoder();
      entryCodingBuffers.set(coder);
      return codec.encodeCopy(entry, dataConfig);
    }

    return coder.encodeCopy(entry, dataConfig);
    finally
    {
      codec.release();
    }
  }



  /**
   * Insert a record into the entry database.
   *
@@ -323,21 +371,17 @@
       throws DatabaseException, DirectoryException
  {
    DatabaseEntry key = id.getDatabaseEntry();
    EntryCoder coder = entryCodingBuffers.get();
    if(coder == null)
    EntryCodec codec = acquireEntryCodec();
    try
    {
      coder = new EntryCoder();
      entryCodingBuffers.set(coder);
      DatabaseEntry data = codec.encodeInternal(entry, dataConfig);
      OperationStatus status = insert(txn, key, data);
      return (status == OperationStatus.SUCCESS);
    }
    DatabaseEntry data = coder.encodeInternal(entry, dataConfig);

    OperationStatus status;
    status = insert(txn, key, data);
    if (status != OperationStatus.SUCCESS)
    finally
    {
      return false;
      codec.release();
    }
    return true;
  }

  /**
@@ -355,21 +399,17 @@
       throws DatabaseException, DirectoryException
  {
    DatabaseEntry key = id.getDatabaseEntry();
    EntryCoder coder = entryCodingBuffers.get();
    if(coder == null)
    EntryCodec codec = acquireEntryCodec();
    try
    {
      coder = new EntryCoder();
      entryCodingBuffers.set(coder);
      DatabaseEntry data = codec.encodeInternal(entry, dataConfig);
      OperationStatus status = put(txn, key, data);
      return (status == OperationStatus.SUCCESS);
    }
    DatabaseEntry data = coder.encodeInternal(entry, dataConfig);

    OperationStatus status;
    status = put(txn, key, data);
    if (status != OperationStatus.SUCCESS)
    finally
    {
      return false;
      codec.release();
    }
    return true;
  }

  /**

 opends/src/server/org/opends/server/core/CoreConfigManager.java

@@ -23,7 +23,7 @@
 *
 *
 *      Copyright 2006-2010 Sun Microsystems, Inc.
 *      Portions copyright 2011 ForgeRock AS.
 *      Portions copyright 2011-2012 ForgeRock AS.
 */
package org.opends.server.core;
import org.opends.messages.Message;
@@ -370,6 +370,9 @@

    DirectoryServer.setMaxPersistentSearchLimit(
        globalConfig.getMaxPsearches());

    DirectoryServer.setMaxInternalBufferSize((int) globalConfig
        .getMaxInternalBufferSize());
  }



 opends/src/server/org/opends/server/core/DirectoryServer.java

@@ -637,6 +637,10 @@
  // mode is 'manual'.
  private WorkflowElementConfigManager workflowElementConfigManager;

  // The maximum size that internal buffers will be allowed to grow to until
  // they are trimmed.
  private int maxInternalBufferSize = DEFAULT_MAX_INTERNAL_BUFFER_SIZE;

  /**
   * The default timeout used to start the server in detach mode.
   */
@@ -9968,5 +9972,38 @@
    }
  }



  /**
   * Sets the threshold capacity beyond which internal cached buffers used for
   * encoding and decoding entries and protocol messages will be trimmed after
   * use.
   *
   * @param maxInternalBufferSize
   *          The threshold capacity beyond which internal cached buffers used
   *          for encoding and decoding entries and protocol messages will be
   *          trimmed after use.
   */
  public static void setMaxInternalBufferSize(int maxInternalBufferSize)
  {
    directoryServer.maxInternalBufferSize = maxInternalBufferSize;
  }



  /**
   * Returns the threshold capacity beyond which internal cached buffers used
   * for encoding and decoding entries and protocol messages will be trimmed
   * after use.
   *
   * @return The threshold capacity beyond which internal cached buffers used
   *         for encoding and decoding entries and protocol messages will be
   *         trimmed after use.
   */
  public static int getMaxInternalBufferSize()
  {
    return directoryServer.maxInternalBufferSize;
  }

}


 opends/src/server/org/opends/server/extensions/SASLByteChannel.java

@@ -147,8 +147,8 @@
          {
            // Avoid extra copy if ByteBuffer is array based.
            wrappedDataBytes = saslContext.wrap(unwrappedData.array(),
                unwrappedData.arrayOffset(), wrapSize);
            unwrappedData.position(unwrappedData.position() + wrapSize);
                unwrappedData.arrayOffset() + unwrappedData.position(),
                wrapSize);
          }
          else
          {
@@ -156,8 +156,8 @@
            unwrappedData.get(sendUnwrappedBytes, 0, wrapSize);
            wrappedDataBytes = saslContext
                .wrap(sendUnwrappedBytes, 0, wrapSize);
            unwrappedData.position(unwrappedData.position() + wrapSize);
          }
          unwrappedData.position(unwrappedData.position() + wrapSize);

          // Encode SASL packet: 4 byte length + wrapped data.
          if (sendWrappedBuffer.capacity() < wrappedDataBytes.length + 4)

 opends/src/server/org/opends/server/protocols/asn1/ASN1.java

@@ -23,16 +23,17 @@
 *
 *
 *      Copyright 2006-2010 Sun Microsystems, Inc.
 *      Portions copyright 2012 ForgeRock AS.
 */
package org.opends.server.protocols.asn1;



import static org.opends.server.util.ServerConstants.*;

import java.io.InputStream;
import java.io.OutputStream;
import java.nio.channels.ReadableByteChannel;
import java.nio.channels.WritableByteChannel;
import java.util.concurrent.locks.ReentrantLock;

import org.opends.server.types.ByteSequence;
import org.opends.server.types.ByteString;
@@ -40,6 +41,7 @@
import org.opends.server.types.ByteSequenceReader;



/**
 * This class contains various static factory methods for creating
 * ASN.1 readers and writers.
@@ -232,9 +234,32 @@
   */
  public static ASN1Writer getWriter(ByteStringBuilder builder)
  {
    return getWriter(builder, DEFAULT_MAX_INTERNAL_BUFFER_SIZE);
  }



  /**
   * Gets an ASN.1 writer whose destination is the provided byte string builder.
   *
   * @param builder
   *          The byte string builder to use.
   * @param maxInternalBufferSize
   *          The threshold capacity beyond which internal cached buffers used
   *          for encoding and decoding protocol messages will be trimmed after
   *          use.
   * @return The new ASN.1 writer.
   */
  public static ASN1Writer getWriter(ByteStringBuilder builder,
      int maxInternalBufferSize)
  {
    if (maxInternalBufferSize <= 0)
    {
      throw new IllegalArgumentException();
    }
    ByteSequenceOutputStream outputStream = new ByteSequenceOutputStream(
        builder);
    return getWriter(outputStream);
        builder, maxInternalBufferSize);
    return getWriter(outputStream, maxInternalBufferSize);
  }


@@ -249,54 +274,31 @@
   */
  public static ASN1Writer getWriter(OutputStream stream)
  {
    return new ASN1OutputStreamWriter(stream);
    return getWriter(stream, DEFAULT_MAX_INTERNAL_BUFFER_SIZE);
  }



  /**
   * Gets an ASN.1 writer whose destination is the provided writable
   * byte channel and which will use a 4KB buffer.
   * <p>
   * The NIO {@code ByteBuffer} will be flushed to the channel
   * automatically when full or when the {@code ASN1Writer.flush()}
   * method is called.
   * Gets an ASN.1 writer whose destination is the provided output
   * stream.
   *
   * @param channel
   *          The writable byte channel.
   * @param writeLock
   *          The write lock to use when flushing to the destination.
   * @param stream
   *          The output stream to use.
   * @param maxInternalBufferSize
   *          The threshold capacity beyond which internal cached buffers used
   *          for encoding and decoding protocol messages will be trimmed after
   *          use.
   * @return The new ASN.1 writer.
   */
  public static ASN1Writer getWriter(WritableByteChannel channel,
                                     ReentrantLock writeLock)
  public static ASN1Writer getWriter(OutputStream stream,
      int maxInternalBufferSize)
  {
    return new ASN1ByteChannelWriter(channel, writeLock, 4096);
  }



  /**
   * Gets an ASN.1 writer whose destination is the provided writable
   * byte channel.
   * <p>
   * The NIO {@code ByteBuffer} will be flushed to the channel
   * automatically when full or when the {@code ASN1Writer.flush()}
   * method is called.
   *
   * @param channel
   *          The writable byte channel.
   * @param writeLock
   *          The write lock to use when flushing to the destination.
   * @param bufferSize
   *          The buffer size to use when writing to the channel.
   * @return The new ASN.1 writer.
   */
  public static ASN1Writer getWriter(WritableByteChannel channel,
                                     ReentrantLock writeLock,
                                     int bufferSize)
  {
    return new ASN1ByteChannelWriter(channel, writeLock, bufferSize);
    if (maxInternalBufferSize <= 0)
    {
      throw new IllegalArgumentException();
    }
    return new ASN1OutputStreamWriter(stream, maxInternalBufferSize);
  }



 opends/src/server/org/opends/server/protocols/asn1/ASN1ByteChannelWriter.java

File was deleted

 opends/src/server/org/opends/server/protocols/asn1/ASN1OutputStreamWriter.java

@@ -23,6 +23,7 @@
 *
 *
 *      Copyright 2006-2009 Sun Microsystems, Inc.
 *      Portions copyright 2012 ForgeRock AS.
 */
package org.opends.server.protocols.asn1;

@@ -48,22 +49,30 @@
  private static final DebugTracer TRACER = getTracer();

  private final OutputStream rootStream;
  private OutputStream out;
  private final ArrayList<ByteSequenceOutputStream> streamStack;
  private final int maxInternalBufferSize;
  private OutputStream out;
  private int stackDepth;



  /**
   * Creates a new ASN.1 output stream reader.
   *
   * @param stream
   *          The underlying output stream.
   * @param maxInternalBufferSize
   *          The threshold capacity beyond which internal cached buffers used
   *          for encoding and decoding protocol messages will be trimmed after
   *          use.
   */
  ASN1OutputStreamWriter(OutputStream stream)
  ASN1OutputStreamWriter(OutputStream stream, int maxInternalBufferSize)
  {
    this.out = stream;
    this.rootStream = stream;
    this.streamStack = new ArrayList<ByteSequenceOutputStream>();
    this.stackDepth = -1;
    this.maxInternalBufferSize = maxInternalBufferSize;
  }

  /**
@@ -470,14 +479,16 @@
    // Make sure we have a cached sub-stream at this depth
    if(stackDepth >= streamStack.size())
    {
      ByteSequenceOutputStream subStream =
          new ByteSequenceOutputStream(new ByteStringBuilder());
      ByteSequenceOutputStream subStream = new ByteSequenceOutputStream(
          new ByteStringBuilder(), maxInternalBufferSize);
      streamStack.add(subStream);
      out = subStream;
    }
    else
    {
      out = streamStack.get(stackDepth);
      ByteSequenceOutputStream childStream = streamStack.get(stackDepth);
      childStream.reset(); // Precaution.
      out = childStream;
    }
/*
    if(debugEnabled())
@@ -535,29 +546,42 @@
  }

  /**
   * Closes this ASN.1 writer and the underlying outputstream. Any unfinished
   * sequences will be ended.
   *
   * @throws IOException if an error occurs while closing the stream.
   * {@inheritDoc}
   */
  public void close() throws IOException {
    while(stackDepth >= 0)
  public void close() throws IOException
  {
    try
    {
      flush();
    }
    finally
    {
      try
      {
        rootStream.close();
      }
      finally
      {
        // Reset for next usage in case where the root stream is reusable (e.g.
        // ByteStringBuilder).
        stackDepth = -1;
        out = rootStream;
      }
    }
  }



  /**
   * {@inheritDoc}
   */
  public void flush() throws IOException
  {
    while (stackDepth >= 0)
    {
      writeEndSequence();
    }
    rootStream.flush();

    streamStack.clear();
    rootStream.close();
  }

  /**
   * Flushes the stream.
   *
   * @throws IOException If an I/O error occurs
   */
  public void flush() throws IOException {
    rootStream.flush();
  }

  /**

 opends/src/server/org/opends/server/protocols/asn1/ByteSequenceOutputStream.java

@@ -23,6 +23,7 @@
 *
 *
 *      Copyright 2006-2008 Sun Microsystems, Inc.
 *      Portions copyright 2012 ForgeRock AS.
 */
package org.opends.server.protocols.asn1;

@@ -36,18 +37,26 @@
 * with the outputstream interface.
 */
final class ByteSequenceOutputStream extends OutputStream {
  private static final int BUFFER_INIT_SIZE = 32;

  private final ByteStringBuilder buffer;
  private final int maxInternalBufferSize;

  /**
   * Creates a new byte string builder output stream.
   *
   * @param buffer
   *          The underlying byte string builder.
   * @param maxInternalBufferSize
   *          The threshold capacity beyond which internal cached buffers used
   *          for encoding and decoding protocol messages will be trimmed after
   *          use.
   */
  ByteSequenceOutputStream(ByteStringBuilder buffer)
  ByteSequenceOutputStream(ByteStringBuilder buffer, int maxInternalBufferSize)
  {
    this.buffer = buffer;
    this.maxInternalBufferSize = Math.max(maxInternalBufferSize,
        BUFFER_INIT_SIZE);
  }

  /**
@@ -98,18 +107,25 @@
  }

  /**
   * Resets this output stream such that the underlying byte string
   * builder is empty.
   * Resets this output stream such that the underlying byte string builder is
   * empty, and also has a capacity which is not too big.
   */
  void reset()
  {
    buffer.clear();
    if (buffer.capacity() > maxInternalBufferSize)
    {
      buffer.clear(BUFFER_INIT_SIZE);
    }
    else
    {
      buffer.clear();
    }
  }

  /**
   * {@inheritDoc}
   */
  public void close() throws IOException {
    buffer.clear();
    reset();
  }
}

 opends/src/server/org/opends/server/protocols/ldap/LDAPClientConnection.java

@@ -31,6 +31,7 @@

import static org.opends.messages.CoreMessages.ERR_ENQUEUE_BIND_IN_PROGRESS;
import static org.opends.messages.ProtocolMessages.*;
import static org.opends.server.core.DirectoryServer.getMaxInternalBufferSize;
import static org.opends.server.loggers.AccessLogger.logDisconnect;
import static org.opends.server.loggers.ErrorLogger.logError;
import static org.opends.server.loggers.debug.DebugLogger.debugEnabled;
@@ -48,11 +49,9 @@
import java.util.Collection;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.atomic.AtomicLong;
import java.util.concurrent.atomic.AtomicReference;
import java.util.concurrent.locks.ReentrantLock;

import org.opends.messages.Message;
import org.opends.messages.MessageBuilder;
@@ -79,10 +78,10 @@
 * connection handler and have its requests decoded by an LDAP request
 * handler.
 */

public class LDAPClientConnection extends ClientConnection implements
public final class LDAPClientConnection extends ClientConnection implements
    TLSCapableConnection
{

  /**
   * A runnable whose task is to close down all IO related channels
   * associated with a client connection after a small delay.
@@ -194,8 +193,8 @@
      if (selector == null)
      {
        // The client connection does not provide a selector, so we'll
        // fall back
        // to a more inefficient way that will work without a selector.
        // fall back to a more inefficient way that will work without a
        // selector.
        while (byteBuffer.hasRemaining()
               && (System.currentTimeMillis() < stopTime))
        {
@@ -278,6 +277,48 @@
   */
  private static final DebugTracer TRACER = getTracer();

  /**
   * Thread local ASN1Writer and buffer.
   */
  private static final class ASN1WriterHolder
  {
    private final ASN1Writer writer;
    private final ByteStringBuilder buffer;
    private final int maxBufferSize;

    private ASN1WriterHolder()
    {
      this.buffer = new ByteStringBuilder();
      this.maxBufferSize = getMaxInternalBufferSize();
      this.writer = ASN1.getWriter(buffer, maxBufferSize);
    }
  }

  // Cached ASN1 writer: a thread can only write to one connection at a time.
  private static final ThreadLocal<ASN1WriterHolder> ASN1_WRITER_CACHE =
      new ThreadLocal<ASN1WriterHolder>()
  {
    /**
     * {@inheritDoc}
     */
    protected ASN1WriterHolder initialValue()
    {
      return new ASN1WriterHolder();
    }
  };

  private ASN1WriterHolder getASN1Writer()
  {
    ASN1WriterHolder holder = ASN1_WRITER_CACHE.get();
    if (holder.maxBufferSize != getMaxInternalBufferSize())
    {
      // Setting has changed, so recreate the holder.
      holder = new ASN1WriterHolder();
      ASN1_WRITER_CACHE.set(holder);
    }
    return holder;
  }

  // The time that the last operation was completed.
  private final AtomicLong lastCompletionTime;

@@ -321,10 +362,6 @@
  // connection.
  private final LDAPConnectionHandler connectionHandler;

  // The reference to the request handler with which this connection is
  // associated.
  private final LDAPRequestHandler requestHandler;

  // The statistics tracker associated with this client connection.
  private final LDAPStatistics statTracker;
  private boolean useNanoTime=false;
@@ -354,17 +391,12 @@
  // client has connected.
  private final String serverAddress;

  // Holds the mapping between the worker thread and ASN1
  // writer to allow for parallel, non-blocking encoding.
  private Map<Thread,ASN1Writer> asn1WriterMap;


  private ASN1ByteChannelReader asn1Reader;

  private final int bufferSize;

  private final RedirectingByteChannel saslChannel;
  private final RedirectingByteChannel tlsChannel;
  private final ReentrantLock writeLock;
  private volatile ConnectionSecurityProvider activeProvider = null;
  private volatile ConnectionSecurityProvider tlsPendingProvider = null;
  private volatile ConnectionSecurityProvider saslPendingProvider = null;
@@ -381,7 +413,7 @@
   * @param  protocol String representing the protocol (LDAP or LDAP+SSL).
   * @throws DirectoryException If SSL initialisation fails.
   */
  public LDAPClientConnection(LDAPConnectionHandler connectionHandler,
  LDAPClientConnection(LDAPConnectionHandler connectionHandler,
      SocketChannel clientChannel, String protocol) throws DirectoryException
  {
    this.connectionHandler = connectionHandler;
@@ -394,7 +426,6 @@
    timeoutClientChannel = new TimeoutWriteByteChannel();
    opsInProgressLock = new Object();
    ldapVersion = 3;
    requestHandler = null;
    lastCompletionTime = new AtomicLong(TimeThread.getTime());
    nextOperationID = new AtomicLong(0);
    connectionValid = true;
@@ -430,9 +461,6 @@
    this.asn1Reader =
        ASN1.getReader(saslChannel, bufferSize, connectionHandler
            .getMaxRequestSize());
    writeLock = new ReentrantLock();

    asn1WriterMap = new ConcurrentHashMap<Thread,ASN1Writer>();

    if (connectionHandler.useSSL())
    {
@@ -477,21 +505,6 @@


  /**
   * Retrieves the request handler that will read requests for this
   * client connection.
   *
   * @return The request handler that will read requests for this client
   *         connection, or <CODE>null</CODE> if none has been assigned
   *         yet.
   */
  public LDAPRequestHandler getRequestHandler()
  {
    return requestHandler;
  }



  /**
   * Retrieves the socket channel that can be used to communicate with
   * the client.
   *
@@ -632,20 +645,6 @@


  /**
   * Retrieves the next operation ID that should be used for this
   * connection.
   *
   * @return The next operation ID that should be used for this
   *         connection.
   */
  public long nextOperationID()
  {
    return nextOperationID.getAndIncrement();
  }



  /**
   * Sends a response to the client based on the information in the
   * provided operation.
   *
@@ -930,22 +929,14 @@
   * @param message
   *          The LDAP message to send to the client.
   */
  public void sendLDAPMessage(LDAPMessage message)
  private void sendLDAPMessage(LDAPMessage message)
  {
    // Get the writer used by this thread.
    Thread currentThread = Thread.currentThread();
    ASN1Writer asn1Writer = asn1WriterMap.get(currentThread);
    // Use a thread local writer.
    final ASN1WriterHolder holder = getASN1Writer();
    try
    {
      if (asn1Writer == null)
      {
        asn1Writer = ASN1.getWriter(saslChannel, writeLock,
                  bufferSize);
        asn1WriterMap.put(currentThread, asn1Writer);
      }

      message.write(asn1Writer);
      asn1Writer.flush();
      message.write(holder.writer);
      holder.buffer.copyTo(saslChannel);

      if (debugEnabled())
      {
@@ -971,7 +962,21 @@
      disconnect(DisconnectReason.SERVER_ERROR, false, null);
      return;
    }
  }
    finally
    {
      // Clear and reset all of the internal buffers ready for the next usage.
      try
      {
        // The ASN1Writer is based on a ByteStringBuilder so closing will cause
        // the internal buffers to be resized if needed.
        holder.writer.close();
      }
      catch (IOException ignored)
      {
        // Unreachable.
      }
    }
 }



@@ -1187,7 +1192,7 @@
   *           client already has reached the maximum allowed concurrent
   *           requests).
   */
  public void addOperationInProgress(AbstractOperation operation)
  private void addOperationInProgress(AbstractOperation operation)
      throws DirectoryException
  {
    int messageID = operation.getMessageID();
@@ -1539,7 +1544,7 @@
   *
   * @return the ASN1 reader for this connection
   */
  protected ASN1ByteChannelReader getASN1Reader()
  ASN1ByteChannelReader getASN1Reader()
  {
    return asn1Reader;
  }
@@ -1552,7 +1557,7 @@
   * @return number of bytes read if this connection is still valid
   *         or negative integer to indicate an error otherwise
   */
  public int processDataRead()
  int processDataRead()
  {
    if (bindOrStartTLSInProgress.get())
    {
@@ -1617,7 +1622,7 @@
   *         error and the client has been disconnected as a result, or
   *         if the client unbound from the server.
   */
  public boolean processLDAPMessage(LDAPMessage message)
  boolean processLDAPMessage(LDAPMessage message)
  {
    if (keepStats)
    {
@@ -2580,7 +2585,7 @@
  /**
   * Enable the provider that is inactive.
   */
  public void enableTLS()
  private void enableTLS()
  {
    activeProvider = tlsPendingProvider;
    tlsChannel.redirect(tlsPendingProvider);
@@ -2595,7 +2600,7 @@
   * @param sslProvider
   *          The provider to set the security provider to.
   */
  public void enableSSL(ConnectionSecurityProvider sslProvider)
  private void enableSSL(ConnectionSecurityProvider sslProvider)
  {
    activeProvider = sslProvider;
    tlsChannel.redirect(sslProvider);
@@ -2606,7 +2611,7 @@
  /**
   * Enable the SASL provider that is currently inactive or pending.
   */
  public void enableSASL()
  private void enableSASL()
  {
    activeProvider = saslPendingProvider;
    saslChannel.redirect(saslPendingProvider);

 opends/src/server/org/opends/server/types/ByteSequence.java

@@ -23,6 +23,7 @@
 *
 *
 *      Copyright 2009 Sun Microsystems, Inc.
 *      Portions copyright 2012 ForgeRock AS.
 */
package org.opends.server.types;

@@ -30,6 +31,7 @@

import java.io.IOException;
import java.io.OutputStream;
import java.nio.channels.WritableByteChannel;



@@ -216,6 +218,21 @@


  /**
   * Copies the entire contents of this byte sequence to the provided
   * {@code WritableByteChannel}.
   *
   * @param channel
   *          The {@code WritableByteChannel} to copy to.
   * @return The number of bytes written, possibly zero
   * @throws IOException
   *           If some other I/O error occurs
   * @see WritableByteChannel#write(java.nio.ByteBuffer)
   */
  int copyTo(WritableByteChannel channel) throws IOException;



  /**
   * Indicates whether the provided byte array sub-sequence is equal
   * to this byte sequence. In order for it to be considered equal,
   * the provided byte array sub-sequence must contain the same bytes

 opends/src/server/org/opends/server/types/ByteString.java

@@ -23,6 +23,7 @@
 *
 *
 *      Copyright 2009 Sun Microsystems, Inc.
 *      Portions copyright 2012 ForgeRock AS.
 */
package org.opends.server.types;

@@ -33,6 +34,8 @@

import java.io.IOException;
import java.io.OutputStream;
import java.nio.ByteBuffer;
import java.nio.channels.WritableByteChannel;

import org.opends.server.loggers.debug.DebugTracer;
import org.opends.server.util.StaticUtils;
@@ -496,6 +499,16 @@
  /**
   * {@inheritDoc}
   */
  public int copyTo(WritableByteChannel channel) throws IOException
  {
    return channel.write(ByteBuffer.wrap(buffer, offset, length));
  }



  /**
   * {@inheritDoc}
   */
  public boolean equals(byte[] b, int offset, int length)
      throws IndexOutOfBoundsException
  {

 opends/src/server/org/opends/server/types/ByteStringBuilder.java

@@ -23,6 +23,7 @@
 *
 *
 *      Copyright 2009 Sun Microsystems, Inc.
 *      Portions copyright 2012 ForgeRock AS.
 */
package org.opends.server.types;

@@ -34,6 +35,8 @@
import java.io.InputStream;
import java.io.OutputStream;
import java.nio.ByteBuffer;
import java.nio.channels.ReadableByteChannel;
import java.nio.channels.WritableByteChannel;
import java.util.zip.DataFormatException;

import org.opends.server.loggers.debug.DebugTracer;
@@ -189,6 +192,16 @@
    /**
     * {@inheritDoc}
     */
    public int copyTo(WritableByteChannel channel) throws IOException
    {
      return channel.write(ByteBuffer.wrap(buffer, subOffset, subLength));
    }



    /**
     * {@inheritDoc}
     */
    public boolean equals(byte[] b, int offset, int length)
        throws IndexOutOfBoundsException
    {
@@ -516,7 +529,7 @@
   *          builder.
   * @param length
   *          The maximum number of bytes to be appended from {@code
   *          buffer}.
   *          stream}.
   * @return The number of bytes read from the input stream, or
   *         {@code -1} if the end of the input stream has been
   *         reached.
@@ -546,6 +559,41 @@


  /**
   * Appends the provided {@code ReadableByteChannel} to this byte string
   * builder.
   *
   * @param channel
   *          The {@code ReadableByteChannel} to be appended to this byte string
   *          builder.
   * @param length
   *          The maximum number of bytes to be appended from {@code channel}.
   * @return The number of bytes read, possibly zero, or {@code -1} if the
   *         channel has reached end-of-stream
   * @throws IOException
   *           If some other I/O error occurs
   * @throws IndexOutOfBoundsException
   *           If {@code length} is less than zero.
   * @see ReadableByteChannel#read(ByteBuffer)
   */
  public int append(ReadableByteChannel channel, int length)
      throws IndexOutOfBoundsException, IOException
  {
    if (length < 0)
    {
      throw new IndexOutOfBoundsException();
    }
    ensureAdditionalCapacity(length);
    int bytesRead = channel.read(ByteBuffer.wrap(buffer, this.length, length));
    if (bytesRead > 0)
    {
      this.length += bytesRead;
    }
    return bytesRead;
  }



  /**
   * Appends the big-endian encoded bytes of the provided short to
   * this byte string builder.
   *
@@ -759,6 +807,19 @@


  /**
   * Returns the current capacity of this byte string builder. The capacity may
   * increase as more data is appended.
   *
   * @return The current capacity of this byte string builder.
   */
  public int capacity()
  {
    return buffer.length;
  }



  /**
   * {@inheritDoc}
   */
  public int compareTo(byte[] b, int offset, int length)
@@ -802,9 +863,40 @@


  /**
   * Sets the length of this byte string builder to zero, and resets the
   * capacity to the specified size.
   * <p>
   * <b>NOTE:</b> if this method is called, then
   * {@code ByteSequenceReader.rewind()} must also be called on any associated
   * byte sequence readers in order for them to remain valid.
   *
   * @param capacity
   *          The new capacity.
   * @return This byte string builder.
   * @throws IllegalArgumentException
   *           If the {@code capacity} is negative.
   * @see #asReader()
   */
  public ByteStringBuilder clear(int capacity) throws IllegalArgumentException
  {
    if (capacity < 0)
    {
      throw new IllegalArgumentException();
    }
    if (capacity != buffer.length)
    {
      buffer = new byte[capacity];
    }
    length = 0;
    return this;
  }



  /**
   * Attempts to compress the data in this buffer into the given
   * buffer. Note that if copmpression was not successful, then the
   * data in the destination buffer should be considered invalid.
   * buffer. Note that if compression was not successful, then the
   * destination buffer will remain unchanged.
   *
   * @param output
   *          The destination buffer of compressed data.
@@ -821,7 +913,7 @@
    output.ensureAdditionalCapacity(length);

    int compressedSize = cryptoManager.compress(buffer, 0, length,
        output.buffer, output.length, output.buffer.length);
        output.buffer, output.length, output.buffer.length - output.length);

    if (compressedSize != -1)
    {
@@ -891,6 +983,16 @@


  /**
   * {@inheritDoc}
   */
  public int copyTo(WritableByteChannel channel) throws IOException
  {
    return channel.write(ByteBuffer.wrap(buffer, 0, length));
  }



  /**
   * Ensures that the specified number of additional bytes will fit in
   * this byte string builder and resizes it if necessary.
   *
@@ -900,12 +1002,11 @@
   */
  public ByteStringBuilder ensureAdditionalCapacity(int size)
  {
    int newCount = this.length + size;
    int newCount = length + size;
    if (newCount > buffer.length)
    {
      byte[] newbuffer =
          new byte[Math.max(buffer.length << 1, newCount)];
      System.arraycopy(buffer, 0, newbuffer, 0, buffer.length);
      byte[] newbuffer = new byte[Math.max(buffer.length << 1, newCount)];
      System.arraycopy(buffer, 0, newbuffer, 0, length);
      buffer = newbuffer;
    }
    return this;
@@ -1008,6 +1109,34 @@


  /**
   * Sets the byte value at the specified index.
   * <p>
   * An index ranges from zero to {@code length() - 1}. The first byte value of
   * the sequence is at index zero, the next at index one, and so on, as for
   * array indexing.
   *
   * @param index
   *          The index of the byte to be set.
   * @param b
   *          The new byte value.
   * @return This byte string builder.
   * @throws IndexOutOfBoundsException
   *           If the index argument is negative or not less than length().
   */
  public ByteStringBuilder setByteAt(int index, byte b)
      throws IndexOutOfBoundsException
  {
    if (index >= length || index < 0)
    {
      throw new IndexOutOfBoundsException();
    }
    buffer[index] = b;
    return this;
  }



  /**
   * Returns a new byte sequence that is a subsequence of this byte
   * sequence.
   * <p>
@@ -1106,9 +1235,8 @@

  /**
   * Attempts to uncompress the data in this buffer into the given
   * destination buffer. Note that if decompression was not
   * successful, then the data in the destination buffer should be
   * considered invalid.
   * destination buffer. Note that if uncompression was not
   * successful, then the destination buffer will remain unchanged.
   *
   * @param output
   *          The destination buffer of compressed data.
@@ -1132,14 +1260,14 @@
      output.ensureAdditionalCapacity(uncompressedSize);

    int decompressResult = cryptoManager.uncompress(buffer, 0, length,
        output.buffer, output.length, output.buffer.length);
        output.buffer, output.length, output.buffer.length - output.length);

    if (decompressResult < 0)
    {
      // The destiation buffer wasn't big enough. Resize and retry.
      // The destination buffer wasn't big enough. Resize and retry.
      output.ensureAdditionalCapacity(-(decompressResult));
      decompressResult = cryptoManager.uncompress(buffer, 0, length,
          output.buffer, output.length, output.buffer.length);
          output.buffer, output.length, output.buffer.length - output.length);
    }

    if (decompressResult >= 0)

 opends/src/server/org/opends/server/util/ServerConstants.java

@@ -23,7 +23,7 @@
 *
 *
 *      Copyright 2006-2010 Sun Microsystems, Inc.
 *      Portions Copyright 2010-2011 ForgeRock AS
 *      Portions Copyright 2010-2012 ForgeRock AS
 */
package org.opends.server.util;

@@ -2039,6 +2039,13 @@


  /**
   * Default maximum size for cached protocol/entry encoding buffers.
   */
  public static final int DEFAULT_MAX_INTERNAL_BUFFER_SIZE = 32 * 1024;



  /**
   * The OID for the attribute type that represents the "objectclass" attribute.
   */
  public static final String OBJECTCLASS_ATTRIBUTE_TYPE_OID = "2.5.4.0";

 opends/tests/unit-tests-testng/src/server/org/opends/server/protocols/asn1/ASN1ByteChannelWriterTestCase.java

File was deleted

 opends/tests/unit-tests-testng/src/server/org/opends/server/protocols/asn1/ASN1OutputStreamWriterTestCase.java

@@ -23,6 +23,7 @@
 *
 *
 *      Copyright 2006-2009 Sun Microsystems, Inc.
 *      Portions copyright 2012 ForgeRock AS.
 */
package org.opends.server.protocols.asn1;

@@ -35,7 +36,7 @@
public class ASN1OutputStreamWriterTestCase extends ASN1WriterTestCase
{
  private ByteArrayOutputStream outStream = new ByteArrayOutputStream();
  private ASN1Writer writer = new ASN1OutputStreamWriter(outStream);
  private ASN1Writer writer = ASN1.getWriter(outStream);

  @Override
  ASN1Writer getWriter()

 opends/tests/unit-tests-testng/src/server/org/opends/server/types/ByteSequenceTest.java

@@ -23,6 +23,7 @@
 *
 *
 *      Copyright 2009 Sun Microsystems, Inc.
 *      Portions copyright 2012 ForgeRock AS.
 */

package org.opends.server.types;
@@ -33,7 +34,10 @@

import java.util.Arrays;
import java.io.ByteArrayOutputStream;
import java.io.IOException;
import java.io.UnsupportedEncodingException;
import java.nio.ByteBuffer;
import java.nio.channels.WritableByteChannel;

/**
 * Abstract test case for the ByteSequence interface.
@@ -124,6 +128,24 @@
  }

  @Test(dataProvider = "byteSequenceProvider")
  public void testCopyToWritableByteChannel(ByteSequence bs, byte[] ba) throws Exception
  {
    final ByteStringBuilder dst = new ByteStringBuilder();
    WritableByteChannel channel = new WritableByteChannel()
    {
      public boolean isOpen() { return true; }
      public void close() throws IOException { }
      public int write(ByteBuffer src) throws IOException
      {
        dst.append(src, src.remaining());
        return dst.length();
      }
    };
    bs.copyTo(channel);
    Assert.assertEquals(bs, dst);
  }

  @Test(dataProvider = "byteSequenceProvider")
  public void testEquals(ByteSequence bs, byte[] ba) throws Exception
  {
    Assert.assertTrue(bs.equals(ByteString.wrap(ba)));

 opends/tests/unit-tests-testng/src/server/org/opends/server/types/ByteStringBuilderTest.java

@@ -23,6 +23,7 @@
 *
 *
 *      Copyright 2009 Sun Microsystems, Inc.
 *      Portions copyright 2012 ForgeRock AS.
 */

package org.opends.server.types;
@@ -153,6 +154,16 @@
    bs.byteAt(0);
  }

  @Test(dataProvider = "builderProvider",
      expectedExceptions = IndexOutOfBoundsException.class)
  public void testClearWithNewCapacity(ByteStringBuilder bs, byte[] ba)
  {
    bs.clear(123);
    Assert.assertEquals(bs.length(), 0);
    Assert.assertEquals(bs.capacity(), 123);
    bs.byteAt(0);
  }

  @Test
  public void testEnsureAdditionalCapacity()
  {
@@ -176,9 +187,9 @@
  {
    ByteStringBuilder bsb = new ByteStringBuilder();
    bsb.append(eightBytes);
    Assert.assertTrue(bsb.getBackingArray().length > 8);
    Assert.assertTrue(bsb.capacity() > 8);
    bsb.trimToSize();
    Assert.assertEquals(bsb.getBackingArray().length, 8);
    Assert.assertEquals(bsb.capacity(), 8);
  }

  @Test(expectedExceptions = IndexOutOfBoundsException.class)
@@ -243,4 +254,14 @@
    ByteArrayInputStream stream = new ByteArrayInputStream(new byte[5]);
    Assert.assertEquals(bsb.append(stream, 10), 5);
  }

  @Test
  public void testSetByteAt() throws Exception
  {
    ByteStringBuilder bsb = new ByteStringBuilder();
    bsb.append(0L);
    Assert.assertEquals(bsb.byteAt(0), 0);
    bsb.setByteAt(0, (byte) 0xff);
    Assert.assertEquals(bsb.byteAt(0), (byte) 0xff);
  }
}