ParallelWorkQueue.java revision 748b8e6dae732515ff8e5af17323cdd6b18d6ad8
/*
* 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
* 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 2006-2010 Sun Microsystems, Inc.
* Portions copyright 2013 ForgeRock AS.
*/
/**
* This class defines a data structure for storing and interacting with the
* Directory Server work queue.
*/
public class ParallelWorkQueue
extends WorkQueue<ParallelWorkQueueCfg>
implements ConfigurationChangeListener<ParallelWorkQueueCfg>
{
/**
* The tracer object for the debug logger.
*/
/**
* 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. */
/**
* The number of operations that have been submitted to the work queue for
* processing.
*/
private AtomicLong opsSubmitted;
/**
* 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 thread number used for the last worker thread that was created. */
private int lastThreadNumber;
/**
* 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. */
/** The lock used to provide threadsafe access for the queue. */
/**
* Creates a new instance of this work queue. All initialization should be
* performed in the <CODE>initializeWorkQueue</CODE> method.
*/
public ParallelWorkQueue()
{
// No implementation should be performed here.
}
/**
* {@inheritDoc}
*/
@Override()
{
shutdownRequested = false;
killThreads = false;
// Register to be notified of any configuration changes.
// Get the necessary configuration from the provided entry.
// Create the actual work queue.
// Create the set of worker threads that should be used to service the
// work queue.
{
new ParallelWorkerThread(this, lastThreadNumber);
t.start();
workerThreads.add(t);
}
// Create and register a monitor provider for the work queue.
try
{
new ParallelWorkQueueMonitor(this);
}
catch (Exception e)
{
if (debugEnabled())
{
}
}
}
/**
* {@inheritDoc}
*/
@Override()
{
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.
for (Operation o : pendingOperations)
{
try
{
// The operation has no chance of responding to the cancel
// request so avoid waiting for a cancel response.
if (o.getCancelResult() == null) {
o.abort(cancelRequest);
}
}
catch (Exception e)
{
if (debugEnabled())
{
}
}
}
// Notify all the worker threads of the shutdown.
for (ParallelWorkerThread t : workerThreads)
{
try
{
t.shutDown();
}
catch (Exception e)
{
if (debugEnabled())
{
}
}
}
}
/**
* 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).
*/
{
if (shutdownRequested)
{
}
}
/**
* 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.
*/
{
}
/**
* 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(
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)
{
synchronized (queueLock)
{
try
{
if (currentThreads > numWorkerThreads)
{
{
}
if (currentThreads <= numWorkerThreads)
{
killThreads = false;
}
return null;
}
}
catch (Exception e)
{
if (debugEnabled())
{
}
}
}
}
{
if (numFailures > MAX_RETRY_COUNT)
{
}
return null;
}
try
{
while (true)
{
}
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)
{
synchronized (queueLock)
{
try
{
if (currentThreads > numWorkerThreads)
{
{
}
if (currentThreads <= numWorkerThreads)
{
killThreads = false;
}
return null;
}
}
catch (Exception e)
{
if (debugEnabled())
{
}
}
}
}
}
else
{
return nextOperation;
}
}
}
catch (Exception e)
{
if (debugEnabled())
{
}
// This should not happen. The only recourse we have is to log a message
// and try again.
}
}
/**
* 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.
*/
{
}
/**
* 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 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()
{
}
/**
* {@inheritDoc}
*/
public boolean isConfigurationChangeAcceptable(
{
return true;
}
/**
* {@inheritDoc}
*/
{
// Apply a change to the number of worker threads if appropriate.
if (newNumThreads != currentThreads)
{
synchronized (queueLock)
{
try
{
if (threadsToAdd > 0)
{
for (int i = 0; i < threadsToAdd; i++)
{
new ParallelWorkerThread(this, lastThreadNumber++);
workerThreads.add(t);
t.start();
}
killThreads = false;
}
else
{
killThreads = true;
}
}
catch (Exception e)
{
if (debugEnabled())
{
}
}
}
}
}
/**
* {@inheritDoc}
*/
@Override()
public boolean isIdle()
{
return false;
}
synchronized (queueLock)
{
for (ParallelWorkerThread t : workerThreads)
{
if (t.isActive())
{
return false;
}
}
return true;
}
}
// Determine the number of worker threads.
{
{
// Automatically choose based on the number of processors.
return value;
}
else
{
return configuration.getNumWorkerThreads();
}
}
}