/*

  * 

  * DO NOT ALTER OR REMOVE COPYRIGHT NOTICES OR THIS HEADER.

  * 

  * Copyright 2007-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.grizzly;

 import com.sun.grizzly.util.WorkerThreadImpl;

 import java.nio.channels.SelectionKey;

 import java.util.LinkedList;

 import java.util.concurrent.Callable;

 import com.sun.grizzly.util.ByteBufferFactory.ByteBufferType;

 import java.util.logging.Level;

 /**

  * Simple Thread Pool based on the wait/notify/synchronized mechanism.

  *

  * @author Jean-Francois Arcand

  */

 public class DefaultPipeline extends LinkedList<Callable>

         implements Pipeline<Callable>{

     /**

      * The number of thread waiting for works.

      */

     protected int waitingThreads = 0;

     /**

      * The maximum number of Thread

      */

     protected int maxThreads = 20;

     /**

      * The minimum numbers of {@link WorkerThreadImpl}

      */

     protected int minThreads = 5;

     /**

      * The minimum numbers of spare {@link WorkerThreadImpl}

      */

     protected int minSpareThreads = 2;

     /**

      * The port used.

      */

     protected int port = 8080;

     /**

      * The number of {@link WorkerThreadImpl}

      */

     protected int threadCount = 0;

     /**

      * The name of this Pipeline

      */

     protected String name = "Grizzly";

     /**

      * The Thread Priority

      */

     protected int priority = Thread.NORM_PRIORITY;

     /**

      * Has the pipeline already started

      */

     protected boolean isStarted = false;

     /**

      * {@link WorkerThreadImpl} 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 initial ByteBuffer size for newly created WorkerThread instances

      */

     protected int initialByteBufferSize = 8192;

     /**

      * The {@link ByteBufferType}

      */

     private ByteBufferType byteBufferType = ByteBufferType.HEAP_VIEW;

     // ------------------------------------------------------- Constructor -----/

     public DefaultPipeline(){

         super();

     }

     public DefaultPipeline(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 DefaultPipeline(int maxThreads, int minThreads, String name,

             int port){        

         this(maxThreads,minThreads,name,port,Thread.NORM_PRIORITY);

     }

     // ------------------------------------------------ Lifecycle ------------/

     /**

      * Init the {@link Pipeline} by initializing the required

      * {@link WorkerThreadImpl}. Default value is {@link #maxThreads}

      */

     public synchronized void initPipeline(){

         if (isStarted){

             return;

         }

         if (minThreads > maxThreads) {

             minThreads = maxThreads;

         }

         workerThreads = new WorkerThreadImpl[maxThreads];

         increaseWorkerThread(minThreads, false);

     }

     /**

      * Start the {@link Pipeline} and all associated

      * {@link WorkerThreadImpl}

      */

     public synchronized void startPipeline(){

         if (!isStarted) {

             for (int i=0; i < minThreads; i++){

                 workerThreads[i].start();

             }

             isStarted = true;

         }

     }

     /**

      * Stop the {@link Pipeline} and all associated

      * {@link WorkerThreadImpl}

      */

     public synchronized void stopPipeline(){

         if (isStarted) {            

             isStarted = false;

             for (int i=0; i < threadCount; i++){

                 workerThreads[i].terminate();

             }

         }

         threadCount = 0;

         waitingThreads = 0;

         notifyAll();

     }

     /**

      * Create new {@link WorkerThreadImpl}. This method must be invoked

      * from a synchronized block.

      * @param increment - how many additional {@link WorkerThreadImpl}

      * objects to add

      * @param startThread - should newly added {@link WorkerThreadImpl}

      * objects be started after creation?

      */

     protected void increaseWorkerThread(int increment, boolean startThread){

         WorkerThreadImpl workerThread;

         int currentCount = threadCount;

         int increaseCount = threadCount + increment;

         for (int i=currentCount; i < increaseCount; i++){

             workerThread = createWorkerThread(

                     name + "WorkerThread-"  + port + "-" + i,

                     initialByteBufferSize);

             workerThread.setByteBufferType(byteBufferType);

             workerThread.setPriority(priority);

             if (startThread)

                 workerThread.start();

             workerThreads[i] = workerThread;

             threadCount++;

         }

     }

     /**

      * Interrupt the {@link Thread} using it thread id

      * @param threadID - id of {@link Thread} to interrupt

      * @return boolean, was Thread interrupted successfully ?

      */

     @SuppressWarnings("empty-statement")

     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){

                         Controller.logger().log(Level.FINE,"interruptThread",t);

                     }

                 }

             }

         }

         return false;

     }

     // ---------------------------------------------------- Queue ------------//

     /**

      * Add a {@link Callable} to the execution queue. If there is thread available,

      * the Callable will be automatically executed. If there is no thread available,

      * the Callable will be queued until a thread becomes free. If the queue

      * reach it's maximum (see {@link #getMaxQueueSize}), the Callable will 

      * be rejected. 

      * 

      * If the Callable is an instance of {@link ContextTask}, the associated

      * {@link SelectionKey} will be cancelled and the {@link Context} 

      * recycled. Finally a {@link PipelineFullException} will be propagated.

      * 

      * @param callable a {@link Callable} to add to this Pipeline

      * @throws {@link PipelineFullException} if Pipeline is full

      */

     public synchronized void execute(Callable callable) throws PipelineFullException {

         int queueSize =  size();

         if (maxQueueSizeInBytes != -1 && maxQueueSizeInBytes <= queueSize){

             Controller.logger().log(Level.WARNING, "All threads are busy and" +

                     " the waiting queue is full: " + maxQueueSizeInBytes 

                     + ". Increase the " +

                     "DefaultPipeline.setMaxQueueSize to remove this message.");

             if (callable instanceof ContextTask){

                 ContextTask ct = (ContextTask) callable;

                 SelectionKey key = ct.getContext().getSelectionKey();

                 // Cancel the key.

                 ct.getContext().getSelectorHandler().getSelectionKeyHandler().cancel(key);

                 ct.recycle();

             }

             throw new PipelineFullException("Processing Queue is full");

         }

         addLast(callable);

         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 {@link Callable} object available in the pipeline.

      * All Threads will synchronize on that method

      * @return {@link Callable}

      */

     public synchronized Callable waitForIoTask() {

         if (size() - waitingThreads <= 0) {

             try {

                 waitingThreads++;

                 wait();

             }  catch (InterruptedException e)  {

                 Thread.currentThread().interrupt();

             }

             waitingThreads--;

         }

         return poll();

     }

     /**

      * Invoked when the {@link SelectionKeyHandler} is about to expire a SelectionKey.

      * @param key - A {@link SelectionKey} to expire

      * @return true if the {@link SelectionKeyHandler} should expire the 

      *         {@link SelectionKey}, false if not.

      */

     public boolean expireKey(SelectionKey key){

         return true;

     }

     /**

      * Return <tt>true</tt> if the size of this {@link LinkedList}

      * minus the current waiting threads is lower than zero.

      */

     @Override

     public boolean isEmpty() {

         return  (size() - getWaitingThread() <= 0);

     }

     // --------------------------------------------------Properties ----------//

     /**

      * Return the number of waiting threads.

      * @return number of waiting threads

      */

     public synchronized int getWaitingThread(){

         return waitingThreads;

     }

     /**

      * Set the number of threads used by this pipeline.

      * @param maxThreads maximum number of threads to use

      */

     public synchronized void setMaxThreads(int maxThreads){

         this.maxThreads = maxThreads;

     }

     /**

      * Return the number of threads used by this pipeline.

      * @return maximum number of threads

      */

     public synchronized int getMaxThreads(){

         return maxThreads;

     }

     /**

      * Return current thread count

      * @return current thread count

      */

     public synchronized int getCurrentThreadCount() {

         return threadCount;

     }

     /**

      * Return the curent number of threads that are currently processing

      * a task.

      * @return current busy thread count

      */

     public synchronized int getCurrentThreadsBusy(){

         return (threadCount - waitingThreads);

     }

     /**

      * Return the maximum spare thread.

      * @return maximum spare thread count

      */

     public synchronized int getMaxSpareThreads() {

         return maxThreads;

     }

     /**

      * Return the minimum spare thread.

      * @return minimum spare thread count

      */

     public synchronized int getMinSpareThreads() {

         return minSpareThreads;

     }

     /**

      * Set the minimum spare thread this {@link Pipeline} can handle.

      * @param minSpareThreads minimum number of spare threads to handle

      */

     public synchronized void setMinSpareThreads(int minSpareThreads) {

         this.minSpareThreads = minSpareThreads;

     }

     /**

      * Set the thread priority of the {@link Pipeline}

      * @param priority thread priority to use

      */

     public synchronized void setPriority(int priority){

         this.priority = priority;

     }

     /**

      * Set the name of this {@link Pipeline}

      * @param name Pipeline name to use

      */

     public synchronized void setName(String name){

         this.name = name;

     }

     /**

      * Return the name of this {@link Pipeline}

      * @return the name of this {@link Pipeline}

      */

     public synchronized String getName(){

         return name+port;

     }

     /**

      * Set the port used by this {@link Pipeline}

      * @param port the port used by this {@link Pipeline}

      */

     public synchronized void setPort(int port){

         this.port = port;

     }

     /**

      * Set the minimum thread this {@link Pipeline} 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 {@link Pipeline} will use when increasing the

      * thread pool

      * @param threadsIncrement amount to increase thread pool by

      */

     public synchronized void setThreadsIncrement(int threadsIncrement){

         this.threadsIncrement = threadsIncrement;

     }

     /**

      * The number of {@link Task} currently queued

      * @return number of queued connections

      */

     public synchronized int getTaskQueuedCount(){

         return size();

     }

     /**

      * Set the maximum pending connection this {@link Pipeline}

      * can handle.

      * @param maxQueueSizeInBytesCount maximum queue size (in bytes) this

      * Pipeline should use

      */

     public synchronized void setMaxQueueSize(int maxQueueSizeInBytesCount){

         this.maxQueueSizeInBytes = maxQueueSizeInBytesCount;

     }

     /**

      * Get the maximum pending connections this {@link Pipeline}

      * can handle.

      * @return maximum queue size (in bytes) this Pipeline is using

      */

     public synchronized int getMaxQueueSize(){

         return maxQueueSizeInBytes;

     }

     /**

      * Get the initial WorkerThreadImpl {@link ByteBuffer} size

      * @return initial WorkerThreadImpl <code>ByteBuffwaitingThreadser</code> size

      */

     public synchronized int getInitialByteBufferSize(){

         return initialByteBufferSize;

     }

     /**

      * Set the initial WorkerThreadImpl {@link ByteBuffer} size

      * @param size initial WorkerThreadImpl {@link ByteBuffer} size

      */

     public synchronized void setInitialByteBufferSize(int size){

         initialByteBufferSize = size;

     }

     /**

      * The {@link ByteBufferTypel} used to create the {@link ByteBuffer}

      * associated with {@link WorkerThreadImpl}s created by this instance.

      * @return The {@link ByteBufferTypel} used to create the {@link ByteBuffer}

      * associated with {@link WorkerThreadImpl}s created by this instance.

      */

     public ByteBufferType getByteBufferType() {

         return byteBufferType;

     }

     /**

      * Set the {@link ByteBufferTypel} to use when creating the

      * {@link ByteBuffer} associated with {@link WorkerThreadImpl}s 

      * created by this instance.

      * @param byteBufferType The ByteBuffer type.

      */

     public void setByteBufferType(ByteBufferType byteBufferType) {

         this.byteBufferType = byteBufferType;

     }

     /**

      * Creates an instance of WorkerThreadImpl.

      *

      * @param name {@link String}

      * @param initialByteBufferSize initial {@link ByteBuffer} size

      */

     protected WorkerThreadImpl createWorkerThread(String name,

                     int initialByteBufferSize) {

         return new WorkerThreadImpl(this, name, initialByteBufferSize);

     }

 }