/*
* DO NOT ALTER OR REMOVE COPYRIGHT NOTICES OR THIS HEADER.
*
* Copyright 1997-2008 Sun Microsystems, Inc. All rights reserved.
*
* The contents of this file are subject to the terms of either the GNU
* General Public License Version 2 only ("GPL") or the Common Development
* and Distribution License("CDDL") (collectively, the "License"). You
* may not use this file except in compliance with the License. You can obtain
* a copy of the License at https://glassfish.dev.java.net/public/CDDL+GPL.html
* or glassfish/bootstrap/legal/LICENSE.txt. See the License for the specific
* language governing permissions and limitations under the License.
*
* When distributing the software, include this License Header Notice in each
* file and include the License file at glassfish/bootstrap/legal/LICENSE.txt.
* Sun designates this particular file as subject to the "Classpath" exception
* as provided by Sun in the GPL Version 2 section of the License file that
* accompanied this code. If applicable, add the following below the License
* Header, with the fields enclosed by brackets [] replaced by your own
* identifying information: "Portions Copyrighted [year]
* [name of copyright owner]"
*
* Contributor(s):
*
* If you wish your version of this file to be governed by only the CDDL or
* only the GPL Version 2, indicate your decision by adding "[Contributor]
* elects to include this software in this distribution under the [CDDL or GPL
* Version 2] license." If you don't indicate a single choice of license, a
* recipient has the option to distribute your version of this file under
* either the CDDL, the GPL Version 2 or to extend the choice of license to
* its licensees as provided above. However, if you add GPL Version 2 code
* and therefore, elected the GPL Version 2 license, then the option applies
* only if the new code is made subject to such option by the copyright
* holder.
*/
package com.sun.enterprise.web.connector.grizzly;
import java.nio.channels.SelectionKey;
/**
* Internal FIFO used by the Worker Threads to pass information
* between <code>Task</code> objects.
*
* @author Charlie Hunt
*/
public class CircularQueuePipeline extends CircularQueue<Task> implements Pipeline {
/**
* The number of thread waiting for a <code>Task</code>
*/
protected int waitingThreads = 0;
/**
* The maximum number of Thread
*/
protected int maxThreads = 20;
/**
* The minimum numbers of <code>WorkerThreadImpl</code>
*/
protected int minThreads = 5;
/**
* The minimum numbers of spare <code>WorkerThreadImpl</code>
*/
protected int minSpareThreads = 2;
/**
* The port used.
*/
protected int port = 8080;
/**
* The number of <code>WorkerThreadImpl</code>
*/
protected int threadCount = 0;
/**
* The name of this Pipeline
*/
protected String name;
/**
* The Thread Priority
*/
protected int priority = Thread.NORM_PRIORITY;
/**
* Has the pipeline already started
*/
protected boolean isStarted = false;
/**
* <code>WorkerThreadImpl</code> amanged by this pipeline.
*/
protected transient WorkerThreadImpl[] workerThreads;
/**
* Maximum pending connection before refusing requests.
*/
protected int maxQueueSizeInBytes = -1;
/**
* The increment number used when adding new thread.
*/
protected int threadsIncrement = 1;
/**
* The request times out during transaction.
*/
protected int threadsTimeout = Constants.DEFAULT_TIMEOUT;
/**
* The <code>PipelineStatistic</code> objects used when gathering statistics.
*/
protected transient PipelineStatistic pipelineStat;
// ------------------------------------------------------- Constructor -----/
public CircularQueuePipeline(){
super();
}
public CircularQueuePipeline(int maxThreads, int minThreads, String name,
int port, int priority){
this.maxThreads = maxThreads;
this.port = port;
this.name = name;
this.minThreads = minThreads;
this.priority = priority;
if ( minThreads < minSpareThreads )
minSpareThreads = minThreads;
}
public CircularQueuePipeline(int maxThreads, int minThreads, String name,
int port){
this(maxThreads,minThreads,name,port,Thread.NORM_PRIORITY);
}
// ------------------------------------------------ Lifecycle ------------/
/**
* Init the <code>Pipeline</code> by initializing the required
* <code>WorkerThreadImpl</code>. Default value is 10
*/
public synchronized void initPipeline(){
if (minThreads > maxThreads) {
minThreads = maxThreads;
}
workerThreads = new WorkerThreadImpl[maxThreads];
increaseWorkerThread(minThreads, false);
}
/**
* Start the <code>Pipeline</code> and all associated
* <code>WorkerThreadImpl</code>
*/
public synchronized void startPipeline(){
if (!isStarted) {
for (int i=0; i < minThreads; i++){
workerThreads[i].start();
}
isStarted = true;
}
}
/**
* Stop the <code>Pipeline</code> and all associated
* <code>WorkerThreadImpl</code>
*/
public synchronized void stopPipeline(){
if (isStarted) {
for (int i=0; i < threadCount; i++){
workerThreads[i].terminate();
}
isStarted = false;
}
notifyAll();
}
/**
* Create new <code>WorkerThreadImpl</code>. This method must be invoked
* from a synchronized block.
*/
protected void increaseWorkerThread(int increment, boolean startThread){
WorkerThreadImpl workerThread;
int currentCount = threadCount;
int increaseCount = threadCount + increment;
for (int i=currentCount; i < increaseCount; i++){
workerThread = new WorkerThreadImpl(this,
name + "WorkerThread-" + port + "-" + i);
workerThread.setPriority(priority);
if (startThread)
workerThread.start();
workerThreads[i] = workerThread;
threadCount++;
}
}
/**
* Interrupt the <code>Thread</code> using it thread id
*/
public synchronized boolean interruptThread(long threadID){
ThreadGroup threadGroup = workerThreads[0].getThreadGroup();
Thread[] threads = new Thread[threadGroup.activeCount()];
threadGroup.enumerate(threads);
for (Thread thread: threads){
if ( thread != null && thread.getId() == threadID ){
if ( Thread.State.RUNNABLE != thread.getState()){
try{
thread.interrupt();
return true;
} catch (Throwable t){
// Swallow all exceptions.
}
}
}
}
return false;
}
// ---------------------------------------------------- Queue ------------//
/**
* Add an object to this pipeline
*/
public synchronized void addTask(Task task) {
int queueSize = size();
if ( maxQueueSizeInBytes != -1 && maxQueueSizeInBytes <= queueSize){
task.cancelTask("Maximum Connections Reached: "
+ maxQueueSizeInBytes + " -- Retry later",
"HTTP/1.1 503 Service Unavailable");
task.getSelectorThread().returnTask(task);
return;
}
addLast(task);
notify();
// Create worker threads if we know we will run out of them
if (threadCount < maxThreads && waitingThreads < (queueSize + 1)){
int left = maxThreads - threadCount;
if (threadsIncrement > left){
threadsIncrement = left;
}
increaseWorkerThread(threadsIncrement,true);
}
}
/**
* Return a <code>Task</code> object available in the pipeline.
* All Threads will synchronize on that method
*/
public synchronized Task getTask() {
if (size() - waitingThreads <= 0) {
try {
waitingThreads++;
wait();
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
}
waitingThreads--;
}
if (pipelineStat != null) {
pipelineStat.gather(size());
}
return poll();
}
/**
* Invoked when the SelectorThread is about to expire a SelectionKey.
* @return true if the SelectorThread should expire the SelectionKey, false
* if not.
*/
public boolean expireKey(SelectionKey key){
return true;
}
/**
* Return <code>true</code> if the size of this <code>CircularQueue</code>
* minus the current waiting threads is lower than zero.
*/
public boolean isEmpty() {
return (size() - waitingThreads <= 0);
}
// --------------------------------------------------Properties ----------//
/**
* Return the number of waiting threads.
*/
public synchronized int getWaitingThread(){
return waitingThreads;
}
/**
* Set the number of threads used by this pipeline.
*/
public synchronized void setMaxThreads(int maxThreads){
this.maxThreads = maxThreads;
}
/**
* Return the number of threads used by this pipeline.
*/
public synchronized int getMaxThreads(){
return maxThreads;
}
public synchronized int getCurrentThreadCount() {
return threadCount;
}
/**
* Return the curent number of threads that are currently processing
* a task.
*/
public synchronized int getCurrentThreadsBusy(){
return (threadCount - waitingThreads);
}
/**
* Return the maximum spare thread.
*/
public synchronized int getMaxSpareThreads() {
return maxThreads;
}
/**
* Return the minimum spare thread.
*/
public synchronized int getMinSpareThreads() {
return minSpareThreads;
}
/**
* Set the minimum space thread this <code>Pipeline</code> can handle.
*/
public synchronized void setMinSpareThreads(int minSpareThreads) {
this.minSpareThreads = minSpareThreads;
}
/**
* Set the thread priority of the <code>Pipeline</code>
*/
public synchronized void setPriority(int priority){
this.priority = priority;
}
/**
* Set the name of this <code>Pipeline</code>
*/
public synchronized void setName(String name){
this.name = name;
}
/**
* Return the name of this <code>Pipeline</code>
* @return the name of this <code>Pipeline</code>
*/
public synchronized String getName(){
return name+port;
}
/**
* Set the port used by this <code>Pipeline</code>
* @param port the port used by this <code>Pipeline</code>
*/
public synchronized void setPort(int port){
this.port = port;
}
/**
* Set the minimum thread this <code>Pipeline</code> will creates
* when initializing.
* @param minThreads the minimum number of threads.
*/
public synchronized void setMinThreads(int minThreads){
this.minThreads = minThreads;
}
@Override
public String toString(){
return "name: " + name + " maxThreads: " + maxThreads
+ " type: " + this.getClass().getName();
}
/**
* Set the number the <code>Pipeline</code> will use when increasing the
* thread pool
*/
public synchronized void setThreadsIncrement(int threadsIncrement){
this.threadsIncrement = threadsIncrement;
}
/**
* Set the timeout value a thread will use to times out the request.
*/
public synchronized void setThreadsTimeout(int threadsTimeout){
this.threadsTimeout = threadsTimeout;
}
/**
* The number of <code>Task</code> currently queued
* @return number of queued connections
*/
public synchronized int getTaskQueuedCount(){
return size();
}
/**
* Set the maximum pending connection this <code>Pipeline</code>
* can handle.
*/
public synchronized void setQueueSizeInBytes(int maxQueueSizeInBytesCount){
this.maxQueueSizeInBytes = maxQueueSizeInBytesCount;
}
/**
* Get the maximum pending connection this <code>Pipeline</code>
* can handle.
*/
public synchronized int getQueueSizeInBytes(){
return maxQueueSizeInBytes;
}
/**
* Set the <code>PipelineStatistic</code> object used
* to gather statistic;
*/
public void setPipelineStatistic(PipelineStatistic pipelineStatistic){
this.pipelineStat = pipelineStatistic;
}
/**
* Return the <code>PipelineStatistic</code> object used
* to gather statistic;
*/
public PipelineStatistic getPipelineStatistic(){
return pipelineStat;
}
}