server/extensions/TraditionalWorkQueue.java

	TraditionalWorkQueue.java revision f4d85fde4c95d5f49f683641815e0463d6166720
/*
 * 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
 *
 *
 *      Portions Copyright 2006-2007 Sun Microsystems, Inc.
 */
package org.opends.server.extensions;


import java.util.ArrayList;
import java.util.Iterator;
import java.util.LinkedList;
import java.util.List;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicLong;
import java.util.concurrent.locks.ReentrantLock;

import org.opends.server.api.ConfigurableComponent;
import org.opends.server.api.WorkQueue;
import org.opends.server.config.ConfigAttribute;
import org.opends.server.config.ConfigEntry;
import org.opends.server.config.ConfigException;
import org.opends.server.config.IntegerConfigAttribute;
import org.opends.server.core.DirectoryServer;
import org.opends.server.core.Operation;
import org.opends.server.monitors.TraditionalWorkQueueMonitor;
import org.opends.server.types.CancelRequest;
import org.opends.server.types.ConfigChangeResult;
import org.opends.server.types.DirectoryException;
import org.opends.server.types.DN;
import org.opends.server.types.ErrorLogCategory;
import org.opends.server.types.ErrorLogSeverity;
import org.opends.server.types.InitializationException;
import org.opends.server.types.ResultCode;

import static org.opends.server.config.ConfigConstants.*;
import static org.opends.server.loggers.debug.DebugLogger.debugCought;
import static org.opends.server.loggers.debug.DebugLogger.debugEnabled;
import org.opends.server.types.DebugLogLevel;
import static org.opends.server.loggers.Error.*;
import static org.opends.server.messages.ConfigMessages.*;
import static org.opends.server.messages.CoreMessages.*;
import static org.opends.server.messages.MessageHandler.*;


/**
 * This class defines a data structure for storing and interacting with the
 * Directory Server work queue.
 */
public class TraditionalWorkQueue
       extends WorkQueue
       implements ConfigurableComponent
{


  /**
   * The maximum number of times to retry getting the next operation from the
   * queue if an unexpected failure occurs.
   */
  private static final int MAX_RETRY_COUNT = 5;


  // The set of worker threads that will be used to process this work queue.
  private ArrayList<TraditionalWorkerThread> workerThreads;

  // The number of operations that have been submitted to the work queue for
  // processing.
  private AtomicLong opsSubmitted;

  // The number of times that an attempt to submit a new request has been
  // rejected because the work queue is already at its maximum capacity.
  private AtomicLong queueFullRejects;

  // Indicates whether one or more of the worker threads needs to be killed at
  // the next convenient opportunity.
  private boolean killThreads;

  // Indicates whether the Directory Server is shutting down.
  private boolean shutdownRequested;

  // The DN of the configuration entry with information to use to configure the
  // work queue.
  private DN configEntryDN;

  // The thread number used for the last worker thread that was created.
  private int lastThreadNumber;

  // The maximum number of pending requests that this work queue will allow
  // before it will start rejecting them.
  private int maxCapacity;

  // The number of worker threads that should be active (or will be shortly if
  // a configuration change has not been completely applied).
  private int numWorkerThreads;

  // The queue that will be used to actually hold the pending operations.
  private LinkedBlockingQueue<Operation> opQueue;

  // The lock used to provide threadsafe access for the queue.
  private ReentrantLock queueLock;


  /**
   * Creates a new instance of this work queue.  All initialization should be
   * performed in the <CODE>initializeWorkQueue</CODE> method.
   */
  public TraditionalWorkQueue()
  {
    // No implementation should be performed here.
  }


  /**
   * {@inheritDoc}
   */
  public void initializeWorkQueue(ConfigEntry configEntry)
         throws ConfigException, InitializationException
  {
    shutdownRequested = false;
    killThreads       = false;
    opsSubmitted      = new AtomicLong(0);
    queueFullRejects  = new AtomicLong(0);
    queueLock         = new ReentrantLock();


    // Get the necessary configuration from the provided entry.
    configEntryDN = configEntry.getDN();

    int msgID = MSGID_CONFIG_WORK_QUEUE_DESCRIPTION_NUM_THREADS;
    IntegerConfigAttribute numThreadsStub =
      new IntegerConfigAttribute(ATTR_NUM_WORKER_THREADS, getMessage(msgID),
                                 true, false, false, true, 1, false, 0,
                                 DEFAULT_NUM_WORKER_THREADS);
    try
    {
      IntegerConfigAttribute numThreadsAttr =
        (IntegerConfigAttribute)
        configEntry.getConfigAttribute(numThreadsStub);
      if (numThreadsAttr == null)
      {
        numWorkerThreads = DEFAULT_NUM_WORKER_THREADS;
      }
      else
      {
        numWorkerThreads = numThreadsAttr.activeIntValue();
        if (numWorkerThreads <= 0)
        {
          //This is not valid.  The number of worker threads must be a positive
          // integer.
          msgID = MSGID_CONFIG_WORK_QUEUE_NUM_THREADS_INVALID_VALUE;
          String message = getMessage(msgID,
                                      String.valueOf(configEntryDN),
                                      numWorkerThreads);
          logError(ErrorLogCategory.CONFIGURATION,
                   ErrorLogSeverity.SEVERE_WARNING, message, msgID);
          numWorkerThreads = DEFAULT_NUM_WORKER_THREADS;
        }
      }
    }
    catch (Exception e)
    {
      if (debugEnabled())
      {
        debugCought(DebugLogLevel.ERROR, e);
      }

      msgID = MSGID_CONFIG_WORK_QUEUE_CANNOT_DETERMINE_NUM_WORKER_THREADS;
      String message = getMessage(msgID, String.valueOf(configEntryDN),
                                  String.valueOf(e));
      logError(ErrorLogCategory.CONFIGURATION, ErrorLogSeverity.SEVERE_ERROR,
               message, msgID);

      numWorkerThreads = DEFAULT_NUM_WORKER_THREADS;
    }


    msgID = MSGID_CONFIG_WORK_QUEUE_DESCRIPTION_MAX_CAPACITY;
    IntegerConfigAttribute capacityStub =
      new IntegerConfigAttribute(ATTR_MAX_WORK_QUEUE_CAPACITY,
                                 getMessage(msgID), true, false, false, true,
                                 0, false, 0,
                                 DEFAULT_MAX_WORK_QUEUE_CAPACITY);
    try
    {
      IntegerConfigAttribute capacityAttr =
        (IntegerConfigAttribute)
        configEntry.getConfigAttribute(capacityStub);
      if (capacityAttr == null)
      {
        maxCapacity = DEFAULT_MAX_WORK_QUEUE_CAPACITY;
      }
      else
      {
        maxCapacity = capacityAttr.activeIntValue();
        if (maxCapacity < 0)
        {
          // This is not valid.  The maximum capacity must be greater than or
          // equal to zero.
          msgID = MSGID_CONFIG_WORK_QUEUE_CAPACITY_INVALID_VALUE;
          String message = getMessage(msgID, String.valueOf(configEntryDN),
                                      maxCapacity);
          logError(ErrorLogCategory.CONFIGURATION,
                   ErrorLogSeverity.SEVERE_WARNING, message, msgID);

          maxCapacity = DEFAULT_MAX_WORK_QUEUE_CAPACITY;
        }
      }
    }
    catch (Exception e)
    {
      if (debugEnabled())
      {
        debugCought(DebugLogLevel.ERROR, e);
      }

      msgID = MSGID_CONFIG_WORK_QUEUE_CANNOT_DETERMINE_QUEUE_CAPACITY;
      String message = getMessage(msgID, String.valueOf(configEntryDN),
                                  String.valueOf(e));
      logError(ErrorLogCategory.CONFIGURATION, ErrorLogSeverity.SEVERE_ERROR,
               message, msgID);

      maxCapacity = DEFAULT_MAX_WORK_QUEUE_CAPACITY;
    }


    // Create the actual work queue.
    if (maxCapacity > 0)
    {
      opQueue = new LinkedBlockingQueue<Operation>(maxCapacity);
    }
    else
    {
      opQueue = new LinkedBlockingQueue<Operation>();
    }


    // Create the set of worker threads that should be used to service the
    // work queue.
    workerThreads = new ArrayList<TraditionalWorkerThread>(numWorkerThreads);
    for (lastThreadNumber = 0; lastThreadNumber < numWorkerThreads;
    lastThreadNumber++)
    {
      TraditionalWorkerThread t =
           new TraditionalWorkerThread(this, lastThreadNumber);
      t.start();
      workerThreads.add(t);
    }


    // Register with the Directory Server as a configurable component.
    DirectoryServer.registerConfigurableComponent(this);


    // Create and register a monitor provider for the work queue.
    try
    {
      TraditionalWorkQueueMonitor monitor =
           new TraditionalWorkQueueMonitor(this);
      monitor.initializeMonitorProvider(configEntry);
      monitor.start();
      DirectoryServer.registerMonitorProvider(monitor);
    }
    catch (Exception e)
    {
      if (debugEnabled())
      {
        debugCought(DebugLogLevel.ERROR, e);
      }

      msgID = MSGID_CONFIG_WORK_QUEUE_CANNOT_CREATE_MONITOR;
      String message = getMessage(msgID, TraditionalWorkQueueMonitor.class,
                                  String.valueOf(e));
      logError(ErrorLogCategory.CORE_SERVER, ErrorLogSeverity.SEVERE_ERROR,
               message, msgID);
    }
  }


  /**
   * {@inheritDoc}
   */
  public void finalizeWorkQueue(String reason)
  {
    shutdownRequested = true;


    // Send responses to any operations in the pending queue to indicate that
    // they won't be processed because the server is shutting down.
    CancelRequest cancelRequest = new CancelRequest(true, reason);
    ArrayList<Operation> pendingOperations = new ArrayList<Operation>();
    opQueue.removeAll(pendingOperations);
    for (Operation o : pendingOperations)
    {
      try
      {
        o.cancel(cancelRequest);
      }
      catch (Exception e)
      {
        if (debugEnabled())
        {
          debugCought(DebugLogLevel.ERROR, e);
        }

        logError(ErrorLogCategory.CORE_SERVER, ErrorLogSeverity.SEVERE_WARNING,
                 MSGID_QUEUE_UNABLE_TO_CANCEL, String.valueOf(o),
                 String.valueOf(e));
      }
    }


    // Notify all the worker threads of the shutdown.
    for (TraditionalWorkerThread t : workerThreads)
    {
      try
      {
        t.shutDown();
      }
      catch (Exception e)
      {
        if (debugEnabled())
        {
          debugCought(DebugLogLevel.ERROR, e);
        }

        logError(ErrorLogCategory.CORE_SERVER, ErrorLogSeverity.SEVERE_WARNING,
                 MSGID_QUEUE_UNABLE_TO_NOTIFY_THREAD, t.getName(),
                 String.valueOf(e));
      }
    }
  }


  /**
   * Indicates whether this work queue has received a request to shut down.
   *
   * @return  <CODE>true</CODE> if the work queue has recieved a request to shut
   *          down, or <CODE>false</CODE> if not.
   */
  public boolean shutdownRequested()
  {
    return shutdownRequested;
  }


  /**
   * Submits an operation to be processed by one of the worker threads
   * associated with this work queue.
   *
   * @param  operation  The operation to be processed.
   *
   * @throws  DirectoryException  If the provided operation is not accepted for
   *                              some reason (e.g., if the server is shutting
   *                              down or the pending operation queue is already
   *                              at its maximum capacity).
   */
  public void submitOperation(Operation operation)
         throws DirectoryException
  {
    if (shutdownRequested)
    {
      int    messageID = MSGID_OP_REJECTED_BY_SHUTDOWN;
      String message   = getMessage(messageID);
      throw new DirectoryException(ResultCode.UNAVAILABLE, message, messageID);
    }

    if (! opQueue.offer(operation))
    {
      queueFullRejects.incrementAndGet();

      int    messageID = MSGID_OP_REJECTED_BY_QUEUE_FULL;
      String message   = getMessage(messageID, maxCapacity);
      throw new DirectoryException(ResultCode.UNAVAILABLE, message, messageID);
    }

    opsSubmitted.incrementAndGet();
  }


  /**
   * Retrieves the next operation that should be processed by one of the worker
   * threads, blocking if necessary until a new request arrives.  This method
   * should only be called by a worker thread associated with this work queue.
   *
   * @param  workerThread  The worker thread that is requesting the operation.
   *
   * @return  The next operation that should be processed, or <CODE>null</CODE>
   *          if the server is shutting down and no more operations will be
   *          processed.
   */
  public Operation nextOperation(TraditionalWorkerThread workerThread)
  {
    return retryNextOperation(workerThread, 0);
  }


  /**
   * Retrieves the next operation that should be processed by one of the worker
   * threads following a previous failure attempt.  A maximum of five
   * consecutive failures will be allowed before returning <CODE>null</CODE>,
   * which will cause the associated thread to exit.
   *
   * @param  workerThread  The worker thread that is requesting the operation.
   * @param  numFailures   The number of consecutive failures that the worker
   *                       thread has experienced so far.  If this gets too
   *                       high, then this method will return <CODE>null</CODE>
   *                       rather than retrying.
   *
   * @return  The next operation that should be processed, or <CODE>null</CODE>
   *          if the server is shutting down and no more operations will be
   *          processed, or if there have been too many consecutive failures.
   */
  private Operation retryNextOperation(TraditionalWorkerThread workerThread,
                                       int numFailures)
  {
    // See if we should kill off this thread.  This could be necessary if the
    // number of worker threads has been decreased with the server online.  If
    // so, then return null and the thread will exit.
    if (killThreads)
    {
      queueLock.lock();

      try
      {
        int currentThreads = workerThreads.size();
        if (currentThreads > numWorkerThreads)
        {
          if (workerThreads.remove(Thread.currentThread()))
          {
            currentThreads--;
          }

          if (currentThreads <= numWorkerThreads)
          {
            killThreads = false;
          }

          workerThread.setStoppedByReducedThreadNumber();
          return null;
        }
      }
      catch (Exception e)
      {
        if (debugEnabled())
        {
          debugCought(DebugLogLevel.ERROR, e);
        }
      }
      finally
      {
        queueLock.unlock();
      }
    }

    if ((shutdownRequested) || (numFailures > MAX_RETRY_COUNT))
    {
      if (numFailures > MAX_RETRY_COUNT)
      {
        int msgID = MSGID_CONFIG_WORK_QUEUE_TOO_MANY_FAILURES;
        String message = getMessage(msgID, Thread.currentThread().getName(),
                                    numFailures, MAX_RETRY_COUNT);
        logError(ErrorLogCategory.CORE_SERVER, ErrorLogSeverity.SEVERE_ERROR,
                 message, msgID);
      }

      return null;
    }

    try
    {
      while (true)
      {
        Operation nextOperation = opQueue.poll(5, TimeUnit.SECONDS);
        if (nextOperation == null)
        {
          // There was no work to do in the specified length of time.  See if
          // we should shutdown, and if not then just check again.
          if (shutdownRequested)
          {
            return null;
          }
          else if (killThreads)
          {
            queueLock.lock();

            try
            {
              int currentThreads = workerThreads.size();
              if (currentThreads > numWorkerThreads)
              {
                if (workerThreads.remove(Thread.currentThread()))
                {
                  currentThreads--;
                }

                if (currentThreads <= numWorkerThreads)
                {
                  killThreads = false;
                }

                workerThread.setStoppedByReducedThreadNumber();
                return null;
              }
            }
            catch (Exception e)
            {
              if (debugEnabled())
              {
                debugCought(DebugLogLevel.ERROR, e);
              }
            }
            finally
            {
              queueLock.unlock();
            }
          }
        }
        else
        {
          return nextOperation;
        }
      }
    }
    catch (InterruptedException ie)
    {
      // This is somewhat expected so don't log.
      //      assert debugException(CLASS_NAME, "retryNextOperation", ie);

      // If this occurs, then the worker thread must have been interrupted for
      // some reason.  This could be because the Directory Server is shutting
      // down, in which case we should return null.
      if (shutdownRequested)
      {
        return null;
      }

      // If we've gotten here, then the worker thread was interrupted for some
      // other reason.  This should not happen, and we need to log a message.
      logError(ErrorLogCategory.CORE_SERVER, ErrorLogSeverity.SEVERE_WARNING,
               MSGID_WORKER_INTERRUPTED_WITHOUT_SHUTDOWN,
               Thread.currentThread().getName(), String.valueOf(ie));
      return retryNextOperation(workerThread, numFailures+1);
    }
    catch (Exception e)
    {
      if (debugEnabled())
      {
        debugCought(DebugLogLevel.ERROR, e);
      }

      // This should not happen.  The only recourse we have is to log a message
      // and try again.
      logError(ErrorLogCategory.CORE_SERVER, ErrorLogSeverity.SEVERE_WARNING,
               MSGID_WORKER_WAITING_UNCAUGHT_EXCEPTION,
               Thread.currentThread().getName(), String.valueOf(e));
      return retryNextOperation(workerThread, numFailures + 1);
    }
  }


  /**
   * Attempts to remove the specified operation from this queue if it has not
   * yet been picked up for processing by one of the worker threads.
   *
   * @param  operation  The operation to remove from the queue.
   *
   * @return  <CODE>true</CODE> if the provided request was present in the queue
   *          and was removed successfully, or <CODE>false</CODE> it not.
   */
  public boolean removeOperation(Operation operation)
  {
    return opQueue.remove(operation);
  }


  /**
   * Retrieves the total number of operations that have been successfully
   * submitted to this work queue for processing since server startup.  This
   * does not include operations that have been rejected for some reason like
   * the queue already at its maximum capacity.
   *
   * @return  The total number of operations that have been successfully
   *          submitted to this work queue since startup.
   */
  public long getOpsSubmitted()
  {
    return opsSubmitted.longValue();
  }


  /**
   * Retrieves the total number of operations that have been rejected because
   * the work queue was already at its maximum capacity.
   *
   * @return  The total number of operations that have been rejected because the
   *          work queue was already at its maximum capacity.
   */
  public long getOpsRejectedDueToQueueFull()
  {
    return queueFullRejects.longValue();
  }


  /**
   * Retrieves the number of pending operations in the queue that have not yet
   * been picked up for processing.  Note that this method is not a
   * constant-time operation and can be relatively inefficient, so it should be
   * used sparingly.
   *
   * @return  The number of pending operations in the queue that have not yet
   *          been picked up for processing.
   */
  public int size()
  {
    return opQueue.size();
  }


  /**
   * Retrieves the DN of the configuration entry with which this component is
   * associated.
   *
   * @return  The DN of the configuration entry with which this component is
   *          associated.
   */
  public DN getConfigurableComponentEntryDN()
  {
    return configEntryDN;
  }


  /**
   * Retrieves the set of configuration attributes that are associated with this
   * configurable component.
   *
   * @return  The set of configuration attributes that are associated with this
   *          configurable component.
   */
  public List<ConfigAttribute> getConfigurationAttributes()
  {
    LinkedList<ConfigAttribute> attrList = new LinkedList<ConfigAttribute>();


    int msgID = MSGID_CONFIG_WORK_QUEUE_DESCRIPTION_NUM_THREADS;
    IntegerConfigAttribute numThreadsAttr =
      new IntegerConfigAttribute(ATTR_NUM_WORKER_THREADS, getMessage(msgID),
                                 true, false, false, true, 1, false, 0,
                                 workerThreads.size());
    attrList.add(numThreadsAttr);


    msgID = MSGID_CONFIG_WORK_QUEUE_DESCRIPTION_MAX_CAPACITY;
    IntegerConfigAttribute capacityAttr =
      new IntegerConfigAttribute(ATTR_MAX_WORK_QUEUE_CAPACITY,
                                 getMessage(msgID), true, false, false, true,
                                 0, false, 0, maxCapacity);
    attrList.add(capacityAttr);


    return attrList;
  }


  /**
   * Indicates whether the provided configuration entry has an acceptable
   * configuration for this component.  If it does not, then detailed
   * information about the problem(s) should be added to the provided list.
   *
   * @param  configEntry          The configuration entry for which to make the
   *                              determination.
   * @param  unacceptableReasons  A list that can be used to hold messages about
   *                              why the provided entry does not have an
   *                              acceptable configuration.
   *
   * @return  <CODE>true</CODE> if the provided entry has an acceptable
   *          configuration for this component, or <CODE>false</CODE> if not.
   */
  public boolean hasAcceptableConfiguration(ConfigEntry configEntry,
                                            List<String> unacceptableReasons)
  {
    boolean configIsAcceptable = true;


    // Check the configuration for the number of worker threads.
    int msgID = MSGID_CONFIG_WORK_QUEUE_DESCRIPTION_NUM_THREADS;
    IntegerConfigAttribute numThreadsStub =
      new IntegerConfigAttribute(ATTR_NUM_WORKER_THREADS, getMessage(msgID),
                                 true, false, false, true, 1, false, 0,
                                 workerThreads.size());
    try
    {
      IntegerConfigAttribute numThreadsAttr =
        (IntegerConfigAttribute)
        configEntry.getConfigAttribute(numThreadsStub);
      if (numThreadsAttr == null)
      {
        // This means that the entry doesn't contain the attribute.  This is
        // fine, since we'll just use the default.
      }
      else
      {
        int numWorkerThreads = numThreadsAttr.activeIntValue();
        if (numWorkerThreads <= 0)
        {
          //This is not valid.  The number of worker threads must be a positive
          // integer.
          msgID = MSGID_CONFIG_WORK_QUEUE_NUM_THREADS_INVALID_VALUE;
          String message = getMessage(msgID,
                                      String.valueOf(configEntryDN),
                                      numWorkerThreads);
          unacceptableReasons.add(message);
          configIsAcceptable = false;
        }
      }
    }
    catch (Exception e)
    {
      if (debugEnabled())
      {
        debugCought(DebugLogLevel.ERROR, e);
      }

      msgID = MSGID_CONFIG_WORK_QUEUE_CANNOT_DETERMINE_NUM_WORKER_THREADS;
      String message = getMessage(msgID, String.valueOf(configEntryDN),
                                  String.valueOf(e));
      unacceptableReasons.add(message);
      configIsAcceptable = false;
    }


    // Check the configuration for the maximum work queue capacity.
    msgID = MSGID_CONFIG_WORK_QUEUE_DESCRIPTION_MAX_CAPACITY;
    IntegerConfigAttribute capacityStub =
      new IntegerConfigAttribute(ATTR_MAX_WORK_QUEUE_CAPACITY,
                                 getMessage(msgID), true, false, false, true,
                                 0, false, 0,
                                 maxCapacity);
    try
    {
      IntegerConfigAttribute capacityAttr =
        (IntegerConfigAttribute)
        configEntry.getConfigAttribute(capacityStub);
      if (capacityAttr == null)
      {
        //This means that the entry doesn't contain the attribute.  This is
        // fine, since we'll just use the default.
      }
      else
      {
        int newMaxCapacity = capacityAttr.activeIntValue();
        if (newMaxCapacity < 0)
        {
          // This is not valid.  The maximum capacity must be greater than or
          // equal to zero.
          msgID = MSGID_CONFIG_WORK_QUEUE_CAPACITY_INVALID_VALUE;
          String message = getMessage(msgID, String.valueOf(configEntryDN),
                                      newMaxCapacity);
          unacceptableReasons.add(message);
          configIsAcceptable = false;
        }
      }
    }
    catch (Exception e)
    {
      if (debugEnabled())
      {
        debugCought(DebugLogLevel.ERROR, e);
      }

      msgID = MSGID_CONFIG_WORK_QUEUE_CANNOT_DETERMINE_QUEUE_CAPACITY;
      String message = getMessage(msgID, String.valueOf(configEntryDN),
                                  String.valueOf(e));
      unacceptableReasons.add(message);
      configIsAcceptable = false;
    }


    return configIsAcceptable;
  }


  /**
   * Makes a best-effort attempt to apply the configuration contained in the
   * provided entry.  Information about the result of this processing should be
   * added to the provided message list.  Information should always be added to
   * this list if a configuration change could not be applied.  If detailed
   * results are requested, then information about the changes applied
   * successfully (and optionally about parameters that were not changed) should
   * also be included.
   *
   * @param  configEntry      The entry containing the new configuration to
   *                          apply for this component.
   * @param  detailedResults  Indicates whether detailed information about the
   *                          processing should be added to the list.
   *
   * @return  Information about the result of the configuration update.
   */
  public ConfigChangeResult applyNewConfiguration(ConfigEntry configEntry,
                                                  boolean detailedResults)
  {
    ArrayList<String> resultMessages = new ArrayList<String>();
    int newNumThreads;
    int newMaxCapacity;


    // Check the configuration for the number of worker threads.
    int msgID = MSGID_CONFIG_WORK_QUEUE_DESCRIPTION_NUM_THREADS;
    IntegerConfigAttribute numThreadsStub =
      new IntegerConfigAttribute(ATTR_NUM_WORKER_THREADS, getMessage(msgID),
                                 true, false, false, true, 1, false, 0,
                                 workerThreads.size());
    try
    {
      IntegerConfigAttribute numThreadsAttr =
        (IntegerConfigAttribute)
        configEntry.getConfigAttribute(numThreadsStub);
      if (numThreadsAttr == null)
      {
        // This means that the entry doesn't contain the attribute.  This is
        // fine, since we'll just use the default.
        newNumThreads = DEFAULT_NUM_WORKER_THREADS;
      }
      else
      {
        newNumThreads = numThreadsAttr.activeIntValue();
        if (newNumThreads <= 0)
        {
          //This is not valid.  The number of worker threads must be a positive
          // integer.  This should never happen since it should be filtered out
          // by the hasAcceptableConfiguration method, but if it does for some
          // reason then handle it.
          msgID = MSGID_CONFIG_WORK_QUEUE_NUM_THREADS_INVALID_VALUE;
          String message = getMessage(msgID,
                                      String.valueOf(configEntryDN),
                                      newNumThreads);
          resultMessages.add(message);
          newNumThreads = DEFAULT_NUM_WORKER_THREADS;
        }
      }
    }
    catch (Exception e)
    {
      if (debugEnabled())
      {
        debugCought(DebugLogLevel.ERROR, e);
      }

      msgID = MSGID_CONFIG_WORK_QUEUE_CANNOT_DETERMINE_NUM_WORKER_THREADS;
      String message = getMessage(msgID, String.valueOf(configEntryDN),
                                  String.valueOf(e));
      resultMessages.add(message);
      newNumThreads = DEFAULT_NUM_WORKER_THREADS;
    }


    // Check the configuration for the maximum work queue capacity.
    msgID = MSGID_CONFIG_WORK_QUEUE_DESCRIPTION_MAX_CAPACITY;
    IntegerConfigAttribute capacityStub =
      new IntegerConfigAttribute(ATTR_MAX_WORK_QUEUE_CAPACITY,
                                 getMessage(msgID), true, false, false, true,
                                 0, false, 0,
                                 maxCapacity);
    try
    {
      IntegerConfigAttribute capacityAttr =
        (IntegerConfigAttribute)
        configEntry.getConfigAttribute(capacityStub);
      if (capacityAttr == null)
      {
        //This means that the entry doesn't contain the attribute.  This is
        // fine, since we'll just use the default.
        newMaxCapacity = DEFAULT_MAX_WORK_QUEUE_CAPACITY;
      }
      else
      {
        newMaxCapacity = capacityAttr.activeIntValue();
        if (newMaxCapacity < 0)
        {
          // This is not valid.  The maximum capacity must be greater than or
          // equal to zero.
          msgID = MSGID_CONFIG_WORK_QUEUE_CAPACITY_INVALID_VALUE;
          String message = getMessage(msgID, String.valueOf(configEntryDN),
                                      newMaxCapacity);
          resultMessages.add(message);
          newMaxCapacity = DEFAULT_MAX_WORK_QUEUE_CAPACITY;
        }
      }
    }
    catch (Exception e)
    {
      if (debugEnabled())
      {
        debugCought(DebugLogLevel.ERROR, e);
      }

      msgID = MSGID_CONFIG_WORK_QUEUE_CANNOT_DETERMINE_QUEUE_CAPACITY;
      String message = getMessage(msgID, String.valueOf(configEntryDN),
                                  String.valueOf(e));
      resultMessages.add(message);
      newMaxCapacity = DEFAULT_MAX_WORK_QUEUE_CAPACITY;
    }


    // Apply a change to the number of worker threads if appropriate.
    int currentThreads = workerThreads.size();
    if (newNumThreads != currentThreads)
    {
      queueLock.lock();

      try
      {
        int threadsToAdd = newNumThreads - currentThreads;
        if (threadsToAdd > 0)
        {
          for (int i=0; i < threadsToAdd; i++)
          {
            TraditionalWorkerThread t =
                 new TraditionalWorkerThread(this, lastThreadNumber++);
            workerThreads.add(t);
            t.start();
          }

          if (detailedResults)
          {
            msgID = MSGID_CONFIG_WORK_QUEUE_CREATED_THREADS;
            String message = getMessage(msgID, threadsToAdd, newNumThreads);
            resultMessages.add(message);
          }

          killThreads = false;
        }
        else
        {
          if (detailedResults)
          {
            msgID = MSGID_CONFIG_WORK_QUEUE_DESTROYING_THREADS;
            String message = getMessage(msgID, Math.abs(threadsToAdd),
                                        newNumThreads);
            resultMessages.add(message);
          }

          killThreads = true;
        }

        numWorkerThreads = newNumThreads;
      }
      catch (Exception e)
      {
        if (debugEnabled())
        {
          debugCought(DebugLogLevel.ERROR, e);
        }
      }
      finally
      {
        queueLock.unlock();
      }
    }


    // Apply a change to the maximum capacity if appropriate.  Since we can't
    // change capacity on the fly, then we'll have to create a new queue and
    // transfer any remaining items into it.  Any thread that is waiting on the
    // original queue will time out after at most a few seconds and further
    // checks will be against the new queue.
    if (newMaxCapacity != maxCapacity)
    {
      queueLock.lock();

      try
      {
        LinkedBlockingQueue<Operation> newOpQueue;
        if (newMaxCapacity > 0)
        {
          newOpQueue = new LinkedBlockingQueue<Operation>(newMaxCapacity);
        }
        else
        {
          newOpQueue = new LinkedBlockingQueue<Operation>();
        }

        LinkedBlockingQueue<Operation> oldOpQueue = opQueue;
        opQueue = newOpQueue;

        LinkedList<Operation> pendingOps = new LinkedList<Operation>();
        oldOpQueue.drainTo(pendingOps);


        // We have to be careful when adding any existing pending operations
        // because the new capacity could be less than what was already
        // backlogged in the previous queue.  If that happens, we may have to
        // loop a few times to get everything in there.
        while (! pendingOps.isEmpty())
        {
          Iterator<Operation> iterator = pendingOps.iterator();
          while (iterator.hasNext())
          {
            Operation o = iterator.next();
            try
            {
              if (newOpQueue.offer(o, 1000, TimeUnit.MILLISECONDS))
              {
                iterator.remove();
              }
            }
            catch (InterruptedException ie)
            {
              if (debugEnabled())
              {
                debugCought(DebugLogLevel.ERROR, ie);
              }
            }
          }
        }

        if (detailedResults)
        {
          msgID = MSGID_CONFIG_WORK_QUEUE_NEW_CAPACITY;
          String message = getMessage(msgID, newMaxCapacity);
          resultMessages.add(message);
        }

        maxCapacity = newMaxCapacity;
      }
      catch (Exception e)
      {
        if (debugEnabled())
        {
          debugCought(DebugLogLevel.ERROR, e);
        }
      }
      finally
      {
        queueLock.unlock();
      }
    }


    return new ConfigChangeResult(ResultCode.SUCCESS, false, resultMessages);
  }


  /**
   * {@inheritDoc}
   */
  public boolean isIdle()
  {
    if (opQueue.size() > 0)
    {
      return false;
    }

    queueLock.lock();

    try
    {
      for (TraditionalWorkerThread t : workerThreads)
      {
        if (t.isActive())
        {
          return false;
        }
      }

      return true;
    }
    finally
    {
      queueLock.unlock();
    }
  }
}