OPENDJ-832 Leverage the work queue for processing requests received on the ...

Jean-Noel Rouvignac

22.28.2013 269a1d06ff820c287bb21a03fa76e3314110516a

OPENDJ-832 Leverage the work queue for processing requests received on the HTTP connection handler



Limited the number of concurrent operations per client connection.



SdkConnectionAdapter.java:
Changed the queueing strategy to BoundedWorkQueueStrategy.

BoundedWorkQueueStrategy.java: ADDED

AbandonOperationWrapper.java, xtendedOperationWrapper.java, UnbindOperationWrapper.java: ADDED

DirectoryServer.java:
Extracted checkCanEnqueueRequest() from enqueueRequest().
Added tryEnqueueRequest().

WorkQueue.java:
Pulled computeNumWorkerThreads() here from sub classes.
Added abstract method trySubmitOperation().
Added abstract getter getNumWorkerThreads().

ParallelWorkQueue.java:
Implemented trySubmitOperation() and getter getNumWorkerThreads().
Renamed getNumWorkerThreads() to computeNumWorkerThreads() + moved it to WorkQueue.

TraditionalWorkQueue.java:
Implemented trySubmitOperation() and getter getNumWorkerThreads().
Renamed getNumWorkerThreads() to computeNumWorkerThreads() + moved it to WorkQueue.
Extacted overloaded method submitOperation(Operation, boolean) from submitOperation(Operation).



ClientConnection.java, JmxClientConnection.java, HTTPClientConnection.java, InternalClientConnection.java, MockClientConnection.java, LDAPClientConnection.java:
Added abstract method isConnectionValid() + implemented it in subclasses.



02-config.ldif, HTTPConnectionHandlerConfiguration.xml:
Added property max-concurrent-ops-per-connection

 opends/resource/schema/02-config.ldif

@@ -3855,7 +3855,8 @@
        ds-cfg-max-blocked-write-time-limit $
        ds-cfg-buffer-size $
        ds-cfg-config-file $
        ds-cfg-authentication-required )
        ds-cfg-authentication-required $
        ds-cfg-max-concurrent-ops-per-connection )
  X-ORIGIN 'OpenDJ Directory Server' )
objectClasses: ( 1.3.6.1.4.1.26027.1.2.14
  NAME 'ds-cfg-entry-cache'

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

@@ -481,4 +481,31 @@
      </ldap:attribute>
    </adm:profile>
  </adm:property>
  <adm:property name="max-concurrent-ops-per-connection">
    <adm:synopsis>
      Specifies the maximum number of internal operations that each
      HTTP client connection can execute concurrently.
    </adm:synopsis>
    <adm:description>
      This property allow to limit the impact that each HTTP request can have on
      the whole server by limiting the number of internal operations that each
      HTTP request can execute concurrently.
      A value of 0 means that no limit is enforced.
    </adm:description>
    <adm:default-behavior>
      <adm:alias>
        <adm:synopsis>
          Let the server decide.
        </adm:synopsis>
      </adm:alias>
    </adm:default-behavior>
    <adm:syntax>
      <adm:integer lower-limit="0"/>
    </adm:syntax>
    <adm:profile name="ldap">
      <ldap:attribute>
        <ldap:name>ds-cfg-max-concurrent-ops-per-connection</ldap:name>
      </ldap:attribute>
    </adm:profile>
  </adm:property>
</adm:managed-object>

 opends/src/admin/messages/HTTPConnectionHandlerCfgDefn.properties

@@ -33,6 +33,9 @@
property.listen-port.description=Only a single port number may be provided.
property.max-blocked-write-time-limit.synopsis=Specifies the maximum length of time that attempts to write data to HTTP clients should be allowed to block.
property.max-blocked-write-time-limit.description=If an attempt to write data to a client takes longer than this length of time, then the client connection is terminated.
property.max-concurrent-ops-per-connection.synopsis=Specifies the maximum number of internal operations that each HTTP client connection can execute concurrently.
property.max-concurrent-ops-per-connection.description=This property allow to limit the impact that each HTTP request can have on the whole server by limiting the number of internal operations that each HTTP request can execute concurrently. A value of 0 means that no limit is enforced.
property.max-concurrent-ops-per-connection.default-behavior.alias.synopsis=Let the server decide.
property.max-request-size.synopsis=Specifies the size in bytes of the largest HTTP request message that will be allowed by the HTTP Connection Handler.
property.max-request-size.description=This can help prevent denial-of-service attacks by clients that indicate they send extremely large requests to the server causing it to attempt to allocate large amounts of memory.
property.ssl-cert-nickname.synopsis=Specifies the nickname (also called the alias) of the certificate that the HTTP Connection Handler should use when performing SSL communication.

 opends/src/server/org/opends/server/api/ClientConnection.java

@@ -55,10 +55,10 @@
import org.opends.server.types.AuthenticationInfo;
import org.opends.server.types.CancelRequest;
import org.opends.server.types.CancelResult;
import org.opends.server.types.DN;
import org.opends.server.types.DebugLogLevel;
import org.opends.server.types.DirectoryException;
import org.opends.server.types.DisconnectReason;
import org.opends.server.types.DN;
import org.opends.server.types.Entry;
import org.opends.server.types.IntermediateResponse;
import org.opends.server.types.Operation;
@@ -419,7 +419,13 @@
   */
  public abstract InetAddress getLocalAddress();


  /**
   * Returns whether the Directory Server believes this connection to be valid
   * and available for communication.
   *
   * @return true if the connection is valid, false otherwise
   */
  public abstract boolean isConnectionValid();

  /**
   * Indicates whether this client connection is currently using a

 opends/src/server/org/opends/server/api/WorkQueue.java

@@ -27,6 +27,9 @@
 */
package org.opends.server.api;

import static org.opends.messages.CoreMessages.*;
import static org.opends.server.loggers.ErrorLogger.*;

import org.opends.messages.Message;
import org.opends.server.admin.std.server.WorkQueueCfg;
import org.opends.server.config.ConfigException;
@@ -106,6 +109,22 @@


  /**
   * Tries to submit an operation to be processed in the server, without
   * blocking.
   *
   * @param operation
   *          The operation to be processed.
   * @return true if the operation could be submitted to the queue, false if the
   *         queue was full
   * @throws DirectoryException
   *           If the provided operation is not accepted for some reason (e.g.,
   *           if the server is shutting down).
   */
  public abstract boolean trySubmitOperation(Operation operation)
      throws DirectoryException;


  /**
   * Indicates whether the work queue is currently processing any
   * requests.  Note that this is a point-in-time determination, and
   * if any component of the server wishes to depend on a quiescent
@@ -119,6 +138,42 @@
  public abstract boolean isIdle();


  /**
   * Return the maximum number of worker threads that can be used by this
   * WorkQueue (The WorkQueue could have a thread pool which adjusts its size).
   *
   * @return the maximum number of worker threads that can be used by this
   *         WorkQueue
   */
  public abstract int getNumWorkerThreads();


  /**
   * Computes the number of worker threads to use by the working queue based on
   * the configured number.
   *
   * @param configuredNumWorkerThreads
   *          the configured number of worker threads to use
   * @return the number of worker threads to use
   */
  protected int computeNumWorkerThreads(Integer configuredNumWorkerThreads)
  {
    if (configuredNumWorkerThreads != null)
    {
      return configuredNumWorkerThreads;
    }
    else
    {
      // Automatically choose based on the number of processors.
      int cpus = Runtime.getRuntime().availableProcessors();
      int value = Math.max(24, cpus * 2);

      Message message = INFO_ERGONOMIC_SIZING_OF_WORKER_THREAD_POOL.get(value);
      logError(message);

      return value;
    }
  }

  /**
   * Waits for the work queue to become idle before returning.  Note

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

New file
@@ -0,0 +1,58 @@
/*
 * CDDL HEADER START
 *
 * The contents of this file are subject to the terms of the
 * Common Development and Distribution License, Version 1.0 only
 * (the "License").  You may not use this file except in compliance
 * with the License.
 *
 * You can obtain a copy of the license at
 * trunk/opends/resource/legal-notices/OpenDS.LICENSE
 * or https://OpenDS.dev.java.net/OpenDS.LICENSE.
 * See the License for the specific language governing permissions
 * and limitations under the License.
 *
 * When distributing Covered Code, include this CDDL HEADER in each
 * file and include the License file at
 * trunk/opends/resource/legal-notices/OpenDS.LICENSE.  If applicable,
 * add the following below this CDDL HEADER, with the fields enclosed
 * by brackets "[]" replaced with your own identifying information:
 *      Portions Copyright [yyyy] [name of copyright owner]
 *
 * CDDL HEADER END
 *
 *
 *      Copyright 2013 ForgeRock AS
 */
package org.opends.server.core;


/**
 * This abstract class wraps/decorates a given abandon operation. This class
 * will be extended by sub-classes to enhance the functionality of the
 * AbandonOperationBasis.
 */
public abstract class AbandonOperationWrapper extends
    OperationWrapper<AbandonOperation> implements AbandonOperation
{

  /**
   * Creates a new abandon operation wrapper based on the provided abandon
   * operation.
   *
   * @param abandon
   *          The abandon operation to wrap
   */
  public AbandonOperationWrapper(AbandonOperation abandon)
  {
    super(abandon);
  }

  /** {@inheritDoc} */
  @Override
  public int getIDToAbandon()
  {
    return getOperation().getIDToAbandon();
  }

}

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

New file
@@ -0,0 +1,246 @@
/*
 * CDDL HEADER START
 *
 * The contents of this file are subject to the terms of the
 * Common Development and Distribution License, Version 1.0 only
 * (the "License").  You may not use this file except in compliance
 * with the License.
 *
 * You can obtain a copy of the license at
 * trunk/opends/resource/legal-notices/OpenDS.LICENSE
 * or https://OpenDS.dev.java.net/OpenDS.LICENSE.
 * See the License for the specific language governing permissions
 * and limitations under the License.
 *
 * When distributing Covered Code, include this CDDL HEADER in each
 * file and include the License file at
 * trunk/opends/resource/legal-notices/OpenDS.LICENSE.  If applicable,
 * add the following below this CDDL HEADER, with the fields enclosed
 * by brackets "[]" replaced with your own identifying information:
 *      Portions Copyright [yyyy] [name of copyright owner]
 *
 * CDDL HEADER END
 *
 *
 *      Copyright 2013 ForgeRock AS
 */
package org.opends.server.core;

import java.util.concurrent.atomic.AtomicInteger;

import org.opends.server.types.DirectoryException;
import org.opends.server.types.Operation;

/**
 * A QueueingStrategy that concurrently enqueues a bounded number of operations
 * to the DirectoryServer work queue. If the maximum number of concurrently
 * enqueued operations has been reached or if the work queue if full, then the
 * operation will be executed on the current thread.
 */
public class BoundedWorkQueueStrategy implements QueueingStrategy
{

  /**
   * The number of concurrently running operations for this
   * BoundedWorkQueueStrategy.
   */
  private final AtomicInteger nbRunningOperations = new AtomicInteger(0);
  /** Maximum number of concurrent operations. 0 means "unlimited". */
  private final int maxNbConcurrentOperations;

  /**
   * Constructor for BoundedWorkQueueStrategy.
   *
   * @param maxNbConcurrentOperations
   *          the maximum number of operations that can be concurrently enqueued
   *          to the DirectoryServer work queue
   */
  public BoundedWorkQueueStrategy(Integer maxNbConcurrentOperations)
  {
    if (maxNbConcurrentOperations != null)
    {
      this.maxNbConcurrentOperations = maxNbConcurrentOperations;
    }
    else
    {
      int cpus = Runtime.getRuntime().availableProcessors();
      int numWorkerThreads =
          DirectoryServer.getWorkQueue().getNumWorkerThreads();
      this.maxNbConcurrentOperations =
          Math.max(cpus, numWorkerThreads * 25 / 100);
    }
  }

  /** {@inheritDoc} */
  @Override
  public void enqueueRequest(final Operation operation)
      throws DirectoryException
  {
    if (!operation.getClientConnection().isConnectionValid())
    {
      // do not bother enqueueing
      return;
    }

    if (maxNbConcurrentOperations == 0)
    { // unlimited concurrent operations
      if (!DirectoryServer.tryEnqueueRequest(operation))
      { // avoid potential deadlocks by running in the current thread
        operation.run();
      }
    }
    else if (nbRunningOperations.getAndIncrement() > maxNbConcurrentOperations
        || !DirectoryServer.tryEnqueueRequest(wrap(operation)))
    { // avoid potential deadlocks by running in the current thread
      try
      {
        operation.run();
      }
      finally
      {
        // only decrement when the operation is run synchronously.
        // Otherwise it'll be decremented twice (once more in the wrapper).
        nbRunningOperations.decrementAndGet();
      }
    }
  }

  private Operation wrap(final Operation operation)
  {
    if (operation instanceof AbandonOperation)
    {
      return new AbandonOperationWrapper((AbandonOperation) operation)
      {
        @Override
        public void run()
        {
          runWrapped(operation);
        }
      };
    }
    else if (operation instanceof AddOperation)
    {
      return new AddOperationWrapper((AddOperation) operation)
      {
        @Override
        public void run()
        {
          runWrapped(operation);
        }
      };
    }
    else if (operation instanceof BindOperation)
    {
      return new BindOperationWrapper((BindOperation) operation)
      {
        @Override
        public void run()
        {
          runWrapped(operation);
        }
      };
    }
    else if (operation instanceof CompareOperation)
    {
      return new CompareOperationWrapper((CompareOperation) operation)
      {
        @Override
        public void run()
        {
          runWrapped(operation);
        }
      };
    }
    else if (operation instanceof DeleteOperation)
    {
      return new DeleteOperationWrapper((DeleteOperation) operation)
      {
        @Override
        public void run()
        {
          runWrapped(operation);
        }
      };
    }
    else if (operation instanceof ExtendedOperation)
    {
      return new ExtendedOperationWrapper((ExtendedOperation) operation)
      {
        @Override
        public void run()
        {
          runWrapped(operation);
        }
      };
    }
    else if (operation instanceof ModifyDNOperation)
    {
      return new ModifyDNOperationWrapper((ModifyDNOperation) operation)
      {
        @Override
        public void run()
        {
          runWrapped(operation);
        }
      };
    }
    else if (operation instanceof ModifyOperation)
    {
      return new ModifyOperationWrapper((ModifyOperation) operation)
      {
        @Override
        public void run()
        {
          runWrapped(operation);
        }
      };
    }
    else if (operation instanceof SearchOperation)
    {
      return new SearchOperationWrapper((SearchOperation) operation)
      {
        @Override
        public void run()
        {
          runWrapped(operation);
        }
      };
    }
    else if (operation instanceof UnbindOperation)
    {
      return new UnbindOperationWrapper((UnbindOperation) operation)
      {
        @Override
        public void run()
        {
          runWrapped(operation);
        }
      };
    }
    else
    {
      throw new RuntimeException(
          "Not implemented for " + operation == null ? null : operation
              .getClass().getName());
    }
  }

  /**
   * Execute the provided operation and decrement the number of currently
   * running operations after it has finished executing.
   *
   * @param the
   *          operation to execute
   */
  private void runWrapped(final Operation operation)
  {
    try
    {
      operation.run();
    }
    finally
    {
      nbRunningOperations.decrementAndGet();
    }
  }
}

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

@@ -37,10 +37,8 @@
import static org.opends.server.schema.SchemaConstants.*;
import static org.opends.server.util.DynamicConstants.*;
import static org.opends.server.util.ServerConstants.*;
import static org.opends.server.util.StaticUtils.getExceptionMessage;
import static org.opends.server.util.StaticUtils.stackTraceToSingleLineString;
import static org.opends.server.util.StaticUtils.toLowerCase;
import static org.opends.server.util.Validator.ensureNotNull;
import static org.opends.server.util.StaticUtils.*;
import static org.opends.server.util.Validator.*;

import java.io.*;
import java.lang.management.ManagementFactory;
@@ -7287,15 +7285,16 @@


  /**
   * Adds the provided operation to the work queue so that it will be processed
   * by one of the worker threads.
   * Runs all the necessary checks prior to adding an operation to the work
   * queue. It throws a DirectoryException if one of the check fails.
   *
   * @param  operation  The operation to be added to the work queue.
   *
   * @throws  DirectoryException  If a problem prevents the operation from being
   *                              added to the queue (e.g., the queue is full).
   * @param operation
   *          The operation to be added to the work queue.
   * @throws DirectoryException
   *           If a check failed preventing the operation from being added to
   *           the queue
   */
  public static void enqueueRequest(Operation operation)
  private static void checkCanEnqueueRequest(Operation operation)
         throws DirectoryException
  {
    ClientConnection clientConnection = operation.getClientConnection();
@@ -7348,7 +7347,6 @@
         break;

      }

    }


@@ -7420,12 +7418,41 @@
          // determination up to the modify operation itself.
      }
    }
  }



  /**
   * Adds the provided operation to the work queue so that it will be processed
   * by one of the worker threads.
   *
   * @param  operation  The operation to be added to the work queue.
   *
   * @throws  DirectoryException  If a problem prevents the operation from being
   *                              added to the queue (e.g., the queue is full).
   */
  public static void enqueueRequest(Operation operation)
      throws DirectoryException
  {
    checkCanEnqueueRequest(operation);
    directoryServer.workQueue.submitOperation(operation);
  }

  /**
   * Tries to add the provided operation to the work queue if not full so that
   * it will be processed by one of the worker threads.
   *
   * @param operation
   *          The operation to be added to the work queue.
   * @return true if the operation could be enqueued, false otherwise
   * @throws DirectoryException
   *           If a problem prevents the operation from being added to the queue
   *           (e.g., the queue is full).
   */
  public static boolean tryEnqueueRequest(Operation operation)
      throws DirectoryException
  {
    checkCanEnqueueRequest(operation);
    return directoryServer.workQueue.trySubmitOperation(operation);
  }


  /**

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

New file
@@ -0,0 +1,94 @@
/*
 * CDDL HEADER START
 *
 * The contents of this file are subject to the terms of the
 * Common Development and Distribution License, Version 1.0 only
 * (the "License").  You may not use this file except in compliance
 * with the License.
 *
 * You can obtain a copy of the license at
 * trunk/opends/resource/legal-notices/OpenDS.LICENSE
 * or https://OpenDS.dev.java.net/OpenDS.LICENSE.
 * See the License for the specific language governing permissions
 * and limitations under the License.
 *
 * When distributing Covered Code, include this CDDL HEADER in each
 * file and include the License file at
 * trunk/opends/resource/legal-notices/OpenDS.LICENSE.  If applicable,
 * add the following below this CDDL HEADER, with the fields enclosed
 * by brackets "[]" replaced with your own identifying information:
 *      Portions Copyright [yyyy] [name of copyright owner]
 *
 * CDDL HEADER END
 *
 *
 *      Copyright 2013 ForgeRock AS
 */
package org.opends.server.core;

import org.opends.server.types.ByteString;

/**
 * This abstract class wraps/decorates a given extended operation. This class
 * will be extended by sub-classes to enhance the functionality of the
 * ExtendedOperationBasis.
 */
public abstract class ExtendedOperationWrapper extends
    OperationWrapper<ExtendedOperation> implements ExtendedOperation
{

  /**
   * Creates a new extended operation wrapper based on the provided extended
   * operation.
   *
   * @param extended
   *          The extended operation to wrap
   */
  public ExtendedOperationWrapper(ExtendedOperation extended)
  {
    super(extended);
  }

  /** {@inheritDoc} */
  @Override
  public String getRequestOID()
  {
    return getOperation().getRequestOID();
  }

  /** {@inheritDoc} */
  @Override
  public String getResponseOID()
  {
    return getOperation().getResponseOID();
  }

  /** {@inheritDoc} */
  @Override
  public ByteString getRequestValue()
  {
    return getOperation().getRequestValue();
  }

  /** {@inheritDoc} */
  @Override
  public ByteString getResponseValue()
  {
    return getOperation().getResponseValue();
  }

  /** {@inheritDoc} */
  @Override
  public void setResponseOID(String responseOID)
  {
    getOperation().setResponseOID(responseOID);
  }

  /** {@inheritDoc} */
  @Override
  public void setResponseValue(ByteString responseValue)
  {
    getOperation().setResponseValue(responseValue);
  }

}

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

New file
@@ -0,0 +1,50 @@
/*
 * CDDL HEADER START
 *
 * The contents of this file are subject to the terms of the
 * Common Development and Distribution License, Version 1.0 only
 * (the "License").  You may not use this file except in compliance
 * with the License.
 *
 * You can obtain a copy of the license at
 * trunk/opends/resource/legal-notices/OpenDS.LICENSE
 * or https://OpenDS.dev.java.net/OpenDS.LICENSE.
 * See the License for the specific language governing permissions
 * and limitations under the License.
 *
 * When distributing Covered Code, include this CDDL HEADER in each
 * file and include the License file at
 * trunk/opends/resource/legal-notices/OpenDS.LICENSE.  If applicable,
 * add the following below this CDDL HEADER, with the fields enclosed
 * by brackets "[]" replaced with your own identifying information:
 *      Portions Copyright [yyyy] [name of copyright owner]
 *
 * CDDL HEADER END
 *
 *
 *      Copyright 2013 ForgeRock AS
 */
package org.opends.server.core;

/**
 * This abstract class wraps/decorates a given unbind operation. This class will
 * be extended by sub-classes to enhance the functionality of the
 * UnbindOperationBasis.
 */
public abstract class UnbindOperationWrapper extends
    OperationWrapper<UnbindOperation> implements UnbindOperation
{

  /**
   * Creates a new unbind operation wrapper based on the provided unbind
   * operation.
   *
   * @param unbind
   *          The unbind operation to wrap
   */
  public UnbindOperationWrapper(UnbindOperation unbind)
  {
    super(unbind);
  }

}

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

@@ -29,6 +29,11 @@



import static org.opends.messages.ConfigMessages.*;
import static org.opends.messages.CoreMessages.*;
import static org.opends.server.loggers.ErrorLogger.*;
import static org.opends.server.loggers.debug.DebugLogger.*;

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.ConcurrentLinkedQueue;
@@ -52,11 +57,6 @@
import org.opends.server.types.Operation;
import org.opends.server.types.ResultCode;

import static org.opends.messages.ConfigMessages.*;
import static org.opends.messages.CoreMessages.*;
import static org.opends.server.loggers.ErrorLogger.*;
import static org.opends.server.loggers.debug.DebugLogger.*;



/**
@@ -146,7 +146,8 @@
    configuration.addParallelChangeListener(this);

    // Get the necessary configuration from the provided entry.
    numWorkerThreads = getNumWorkerThreads(configuration);
    numWorkerThreads =
        computeNumWorkerThreads(configuration.getNumWorkerThreads());

    // Create the actual work queue.
    opQueue = new ConcurrentLinkedQueue<Operation>();
@@ -286,6 +287,14 @@
    opsSubmitted.incrementAndGet();
  }

  /** {@inheritDoc} */
  @Override
  public boolean trySubmitOperation(Operation operation)
      throws DirectoryException
  {
    submitOperation(operation);
    return true;
  }


  /**
@@ -515,7 +524,8 @@
                                 ParallelWorkQueueCfg configuration)
  {
    ArrayList<Message> resultMessages = new ArrayList<Message>();
    int newNumThreads  = getNumWorkerThreads(configuration);
    int newNumThreads =
        computeNumWorkerThreads(configuration.getNumWorkerThreads());

    // Apply a change to the number of worker threads if appropriate.
    int currentThreads = workerThreads.size();
@@ -583,25 +593,14 @@
    }
  }



  // Determine the number of worker threads.
  private int getNumWorkerThreads(ParallelWorkQueueCfg configuration)
  /**
   * Return the number of worker threads used by this WorkQueue.
   *
   * @return the number of worker threads used by this WorkQueue
   */
  @Override
  public int getNumWorkerThreads()
  {
    if (configuration.getNumWorkerThreads() == null)
    {
      // Automatically choose based on the number of processors.
      int cpus = Runtime.getRuntime().availableProcessors();
      int value = Math.max(24, cpus * 2);

      Message message = INFO_ERGONOMIC_SIZING_OF_WORKER_THREAD_POOL.get(value);
      logError(message);

      return value;
    }
    else
    {
      return configuration.getNumWorkerThreads();
    }
    return this.numWorkerThreads;
  }
}

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

@@ -51,7 +51,13 @@
import org.opends.server.core.DirectoryServer;
import org.opends.server.loggers.debug.DebugTracer;
import org.opends.server.monitors.TraditionalWorkQueueMonitor;
import org.opends.server.types.*;
import org.opends.server.types.CancelRequest;
import org.opends.server.types.ConfigChangeResult;
import org.opends.server.types.DebugLogLevel;
import org.opends.server.types.DirectoryException;
import org.opends.server.types.InitializationException;
import org.opends.server.types.Operation;
import org.opends.server.types.ResultCode;



@@ -176,7 +182,8 @@
      configuration.addTraditionalChangeListener(this);

      // Get the necessary configuration from the provided entry.
      numWorkerThreads = getNumWorkerThreads(configuration);
      numWorkerThreads =
          computeNumWorkerThreads(configuration.getNumWorkerThreads());
      maxCapacity = configuration.getMaxWorkQueueCapacity();

      // Create the actual work queue.
@@ -334,6 +341,32 @@
  @Override
  public void submitOperation(Operation operation) throws DirectoryException
  {
    submitOperation(operation, isBlocking);
  }

  /** {@inheritDoc} */
  @Override
  public boolean trySubmitOperation(Operation operation)
      throws DirectoryException
  {
    try
    {
      submitOperation(operation, false);
      return true;
    }
    catch (DirectoryException e)
    {
      if (ResultCode.BUSY == e.getResultCode())
      {
        return false;
      }
      throw e;
    }
  }

  private void submitOperation(Operation operation,
      boolean blockEnqueuingWhenFull) throws DirectoryException
  {
    queueReadLock.lock();
    try
    {
@@ -343,7 +376,7 @@
        throw new DirectoryException(ResultCode.UNAVAILABLE, message);
      }

      if (isBlocking)
      if (blockEnqueuingWhenFull)
      {
        try
        {
@@ -659,7 +692,8 @@
      TraditionalWorkQueueCfg configuration)
  {
    ArrayList<Message> resultMessages = new ArrayList<Message>();
    int newNumThreads = getNumWorkerThreads(configuration);
    int newNumThreads =
        computeNumWorkerThreads(configuration.getNumWorkerThreads());
    int newMaxCapacity = configuration.getMaxWorkQueueCapacity();

    // Apply a change to the number of worker threads if appropriate.
@@ -808,25 +842,14 @@
    }
  }



  // Determine the number of worker threads.
  private int getNumWorkerThreads(TraditionalWorkQueueCfg configuration)
  /**
   * Return the number of worker threads used by this WorkQueue.
   *
   * @return the number of worker threads used by this WorkQueue
   */
  @Override
  public int getNumWorkerThreads()
  {
    if (configuration.getNumWorkerThreads() == null)
    {
      // Automatically choose based on the number of processors.
      int cpus = Runtime.getRuntime().availableProcessors();
      int value = Math.max(24, cpus * 2);

      Message message = INFO_ERGONOMIC_SIZING_OF_WORKER_THREAD_POOL.get(value);
      logError(message);

      return value;
    }
    else
    {
      return configuration.getNumWorkerThreads();
    }
    return this.numWorkerThreads;
  }
}

 opends/src/server/org/opends/server/protocols/http/SdkConnectionAdapter.java

@@ -59,6 +59,7 @@
import org.opends.server.core.AbandonOperationBasis;
import org.opends.server.core.AddOperationBasis;
import org.opends.server.core.BindOperationBasis;
import org.opends.server.core.BoundedWorkQueueStrategy;
import org.opends.server.core.CompareOperationBasis;
import org.opends.server.core.DeleteOperationBasis;
import org.opends.server.core.ExtendedOperationBasis;
@@ -67,7 +68,6 @@
import org.opends.server.core.QueueingStrategy;
import org.opends.server.core.SearchOperationBasis;
import org.opends.server.core.UnbindOperationBasis;
import org.opends.server.core.WorkQueueStrategy;
import org.opends.server.loggers.debug.DebugTracer;
import org.opends.server.types.AuthenticationInfo;
import org.opends.server.types.ByteString;
@@ -98,7 +98,7 @@
  private AtomicInteger nextMessageID = new AtomicInteger(0);

  /** The queueing strategy used for this connection. */
  private QueueingStrategy queueingStrategy = new WorkQueueStrategy();
  private final QueueingStrategy queueingStrategy;

  /**
   * Whether this connection has been closed by calling {@link #close()} or
@@ -115,6 +115,9 @@
  public SdkConnectionAdapter(HTTPClientConnection clientConnection)
  {
    this.clientConnection = clientConnection;
    this.queueingStrategy =
        new BoundedWorkQueueStrategy(clientConnection.getConnectionHandler()
            .getCurrentConfig().getMaxConcurrentOpsPerConnection());
  }

  @SuppressWarnings({ "rawtypes", "unchecked" })
@@ -269,6 +272,16 @@
    return enqueueOperation(operation, resultHandler);
  }

  /**
   * Return the queueing strategy used by this connection.
   *
   * @return The queueing strategy used by this connection
   */
  public QueueingStrategy getQueueingStrategy()
  {
    return queueingStrategy;
  }

  /** {@inheritDoc} */
  @Override
  public boolean isClosed()

 opends/src/server/org/opends/server/protocols/internal/InternalClientConnection.java

@@ -617,7 +617,13 @@
    // time limit for internal client connections.
  }


  /** {@inheritDoc} */
  @Override
  public boolean isConnectionValid()
  {
    // This connection is always valid
    return true;
  }

  /**
   * Indicates whether this client connection is currently using a

 opends/src/server/org/opends/server/protocols/jmx/JmxClientConnection.java

@@ -88,7 +88,7 @@
  /**
   * Indicate that the disconnect process is started.
   */
  private Boolean disconnectStarted = false;
  private boolean disconnectStarted = false;


  /**
@@ -357,7 +357,12 @@
    return null;
  }


  /** {@inheritDoc} */
  @Override
  public boolean isConnectionValid()
  {
    return !disconnectStarted;
  }

  /**
   * Indicates whether this client connection is currently using a secure

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

@@ -664,7 +664,12 @@
    return clientChannel.socket().getLocalAddress();
  }


  /** {@inheritDoc} */
  @Override
  public boolean isConnectionValid()
  {
    return this.connectionValid;
  }

  /**
   * Indicates whether this client connection is currently using a

 opends/tests/unit-tests-testng/src/server/org/opends/server/core/networkgroups/MockClientConnection.java

@@ -23,6 +23,7 @@
 *
 *
 *      Copyright 2009 Sun Microsystems, Inc.
 *      Portions Copyright 2013 ForgeRock AS
 */
package org.opends.server.core.networkgroups;

@@ -56,6 +57,7 @@
/**
 * A mock connection for connection criteria testing.
 */
@SuppressWarnings("javadoc")
public final class MockClientConnection extends ClientConnection
{
  private final int clientPort;
@@ -68,7 +70,7 @@

  /**
   * Creates a new mock client connection.
   *
   * 
   * @param clientPort
   *          The client port.
   * @param isSecure
@@ -76,7 +78,7 @@
   * @param bindDN
   *          The client bind DN.
   * @param authMethod
   *          The client authentication mathod.
   *          The client authentication method.
   * @throws Exception
   *           If an unexpected exception occurred.
   */
@@ -271,7 +273,12 @@
    return 0;
  }


  @Override
  public boolean isConnectionValid()
  {
    // This connection is always valid
    return true;
  }

  @Override
  public boolean isSecure()