RC9/qpid/java/broker/src/main/java/org/apache/qpid/server/transport/ThreadPoolFilter.java - qpid - Git at Google

 /*
  *
  * Licensed to the Apache Software Foundation (ASF) under one
  * or more contributor license agreements.  See the NOTICE file
  * distributed with this work for additional information
  * regarding copyright ownership.  The ASF licenses this file
  * to you under the Apache License, Version 2.0 (the
  * "License"); you may not use this file except in compliance
  * with the License.  You may obtain a copy of the License at
  *
  *   http://www.apache.org/licenses/LICENSE-2.0
  *
  * Unless required by applicable law or agreed to in writing,
  * software distributed under the License is distributed on an
  * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
  * KIND, either express or implied.  See the License for the
  * specific language governing permissions and limitations
  * under the License.
  *
  */
 package org.apache.qpid.server.transport;

 import java.util.ArrayList;
 import java.util.IdentityHashMap;
 import java.util.Iterator;
 import java.util.List;
 import java.util.Map;
 import java.util.Set;

 import org.apache.mina.common.IdleStatus;
 import org.apache.mina.common.IoFilterAdapter;
 import org.apache.mina.common.IoHandler;
 import org.apache.mina.common.IoSession;
 import org.apache.mina.util.BlockingQueue;
 import org.apache.mina.util.ByteBufferUtil;
 import org.apache.mina.util.IdentityHashSet;
 import org.apache.mina.util.Queue;
 import org.apache.mina.util.Stack;

 /**
  * A Thread-pooling filter.  This filter forwards {@link IoHandler} events
  * to its thread pool.
  * <p/>
  * This is an implementation of
  * <a href="http://deuce.doc.wustl.edu/doc/pspdfs/lf.pdf">Leader/Followers
  * thread pool</a> by Douglas C. Schmidt et al.
  */
 public class ThreadPoolFilter extends IoFilterAdapter
 {
     /**
      * Default maximum size of thread pool (2G).
      */
     public static final int DEFAULT_MAXIMUM_POOL_SIZE = Integer.MAX_VALUE;

     /**
      * Default keep-alive time of thread pool (1 min).
      */
     public static final int DEFAULT_KEEP_ALIVE_TIME = 60 * 1000;

     /**
      * A queue which contains {@link Integer}s which represents reusable
      * thread IDs.  {@link Worker} first checks this queue and then
      * uses {@link #threadId} when no reusable thread ID is available.
      */
     private static final Queue threadIdReuseQueue = new Queue();
     private static int threadId = 0;

     private static int acquireThreadId()
     {
         synchronized (threadIdReuseQueue)
         {
             Integer id = (Integer) threadIdReuseQueue.pop();
             if (id == null)
             {
                 return ++ threadId;
             }
             else
             {
                 return id.intValue();
             }
         }
     }

     private static void releaseThreadId(int id)
     {
         synchronized (threadIdReuseQueue)
         {
             threadIdReuseQueue.push(new Integer(id));
         }
     }

     private final String threadNamePrefix;
     private final Map buffers = new IdentityHashMap();
     private final BlockingQueue unfetchedSessionBuffers = new BlockingQueue();
     private final Set allSessionBuffers = new IdentityHashSet();

     private Worker leader;
     private final Stack followers = new Stack();
     private final Set allWorkers = new IdentityHashSet();

     private int maximumPoolSize = DEFAULT_MAXIMUM_POOL_SIZE;
     private int keepAliveTime = DEFAULT_KEEP_ALIVE_TIME;

     private boolean shuttingDown;

     private int poolSize;
     private final Object poolSizeLock = new Object();

     /**
      * Creates a new instance of this filter with default thread pool settings.
      */
     public ThreadPoolFilter()
     {
         this("IoThreadPool");
     }

     /**
      * Creates a new instance of this filter with the specified thread name prefix
      * and other default settings.
      *
      * @param threadNamePrefix the prefix of the thread names this pool will create.
      */
     public ThreadPoolFilter(String threadNamePrefix)
     {
         if (threadNamePrefix == null)
         {
             throw new NullPointerException("threadNamePrefix");
         }
         threadNamePrefix = threadNamePrefix.trim();
         if (threadNamePrefix.length() == 0)
         {
             throw new IllegalArgumentException("threadNamePrefix is empty.");
         }
         this.threadNamePrefix = threadNamePrefix;
     }

     public String getThreadNamePrefix()
     {
         return threadNamePrefix;
     }

     public int getPoolSize()
     {
         synchronized (poolSizeLock)
         {
             return poolSize;
         }
     }

     public int getMaximumPoolSize()
     {
         return maximumPoolSize;
     }

     public int getKeepAliveTime()
     {
         return keepAliveTime;
     }

     public void setMaximumPoolSize(int maximumPoolSize)
     {
         if (maximumPoolSize <= 0)
         {
             throw new IllegalArgumentException();
         }
         this.maximumPoolSize = maximumPoolSize;
     }

     public void setKeepAliveTime(int keepAliveTime)
     {
         this.keepAliveTime = keepAliveTime;
     }

     public void init()
     {
         shuttingDown = false;
         leader = new Worker();
         leader.start();
         leader.lead();
     }

     public void destroy()
     {
         shuttingDown = true;
         int expectedPoolSize = 0;
         while (getPoolSize() != expectedPoolSize)
         {
             List allWorkers;
             synchronized (poolSizeLock)
             {
                 allWorkers = new ArrayList(this.allWorkers);
             }

             // You may not interrupt the current thread.
             if (allWorkers.remove(Thread.currentThread()))
             {
                 expectedPoolSize = 1;
             }

             for (Iterator i = allWorkers.iterator(); i.hasNext();)
             {
                 Worker worker = (Worker) i.next();
                 while (worker.isAlive())
                 {
                     worker.interrupt();
                     try
                     {
                         // This timeout will help us from
                         // infinite lock-up and interrupt workers again.
                         worker.join(100);
                     }
                     catch (InterruptedException e)
                     {
                     }
                 }
             }
         }

         this.allSessionBuffers.clear();
         this.unfetchedSessionBuffers.clear();
         this.buffers.clear();
         this.followers.clear();
         this.leader = null;
     }

     private void increasePoolSize(Worker worker)
     {
         synchronized (poolSizeLock)
         {
             poolSize++;
             allWorkers.add(worker);
         }
     }

     private void decreasePoolSize(Worker worker)
     {
         synchronized (poolSizeLock)
         {
             poolSize--;
             allWorkers.remove(worker);
         }
     }

     private void fireEvent(NextFilter nextFilter, IoSession session,
                            EventType type, Object data)
     {
         final BlockingQueue unfetchedSessionBuffers = this.unfetchedSessionBuffers;
         final Set allSessionBuffers = this.allSessionBuffers;
         final Event event = new Event(type, nextFilter, data);

         synchronized (unfetchedSessionBuffers)
         {
             final SessionBuffer buf = getSessionBuffer(session);
             final Queue eventQueue = buf.eventQueue;

             synchronized (buf)
             {
                 eventQueue.push(event);
             }

             if (!allSessionBuffers.contains(buf))
             {
                 allSessionBuffers.add(buf);
                 unfetchedSessionBuffers.push(buf);
             }
         }
     }

     /**
      * Implement this method to fetch (or pop) a {@link SessionBuffer} from
      * the given <tt>unfetchedSessionBuffers</tt>.  The default implementation
      * simply pops the buffer from it.  You could prioritize the fetch order.
      *
      * @return A non-null {@link SessionBuffer}
      */
     protected SessionBuffer fetchSessionBuffer(Queue unfetchedSessionBuffers)
     {
         return (SessionBuffer) unfetchedSessionBuffers.pop();
     }

     private SessionBuffer getSessionBuffer(IoSession session)
     {
         final Map buffers = this.buffers;
         SessionBuffer buf = (SessionBuffer) buffers.get(session);
         if (buf == null)
         {
             synchronized (buffers)
             {
                 buf = (SessionBuffer) buffers.get(session);
                 if (buf == null)
                 {
                     buf = new SessionBuffer(session);
                     buffers.put(session, buf);
                 }
             }
         }
         return buf;
     }

     private void removeSessionBuffer(SessionBuffer buf)
     {
         final Map buffers = this.buffers;
         final IoSession session = buf.session;
         synchronized (buffers)
         {
             buffers.remove(session);
         }
     }

     protected static class SessionBuffer
     {
         private final IoSession session;

         private final Queue eventQueue = new Queue();

         private SessionBuffer(IoSession session)
         {
             this.session = session;
         }

         public IoSession getSession()
         {
             return session;
         }

         public Queue getEventQueue()
         {
             return eventQueue;
         }
     }

     private class Worker extends Thread
     {
         private final int id;
         private final Object promotionLock = new Object();
         private boolean dead;

         private Worker()
         {
             int id = acquireThreadId();
             this.id = id;
             this.setName(threadNamePrefix + '-' + id);
             increasePoolSize(this);
         }

         public boolean lead()
         {
             final Object promotionLock = this.promotionLock;
             synchronized (promotionLock)
             {
                 if (dead)
                 {
                     return false;
                 }

                 leader = this;
                 promotionLock.notify();
             }

             return true;
         }

         public void run()
         {
             for (; ;)
             {
                 if (!waitForPromotion())
                 {
                     break;
                 }

                 SessionBuffer buf = fetchBuffer();
                 giveUpLead();
                 if (buf == null)
                 {
                     break;
                 }

                 processEvents(buf);
                 follow();
                 releaseBuffer(buf);
             }

             decreasePoolSize(this);
             releaseThreadId(id);
         }

         private SessionBuffer fetchBuffer()
         {
             BlockingQueue unfetchedSessionBuffers = ThreadPoolFilter.this.unfetchedSessionBuffers;
             synchronized (unfetchedSessionBuffers)
             {
                 while (!shuttingDown)
                 {
                     try
                     {
                         unfetchedSessionBuffers.waitForNewItem();
                     }
                     catch (InterruptedException e)
                     {
                         continue;
                     }

                     return ThreadPoolFilter.this.fetchSessionBuffer(unfetchedSessionBuffers);
                 }
             }

             return null;
         }

         private void processEvents(SessionBuffer buf)
         {
             final IoSession session = buf.session;
             final Queue eventQueue = buf.eventQueue;
             for (; ;)
             {
                 Event event;
                 synchronized (buf)
                 {
                     event = (Event) eventQueue.pop();
                     if (event == null)
                     {
                         break;
                     }
                 }
                 processEvent(event.getNextFilter(), session,
                              event.getType(), event.getData());
             }
         }

         private void follow()
         {
             final Object promotionLock = this.promotionLock;
             final Stack followers = ThreadPoolFilter.this.followers;
             synchronized (promotionLock)
             {
                 if (this != leader)
                 {
                     synchronized (followers)
                     {
                         followers.push(this);
                     }
                 }
             }
         }

         private void releaseBuffer(SessionBuffer buf)
         {
             final BlockingQueue unfetchedSessionBuffers = ThreadPoolFilter.this.unfetchedSessionBuffers;
             final Set allSessionBuffers = ThreadPoolFilter.this.allSessionBuffers;
             final Queue eventQueue = buf.eventQueue;

             synchronized (unfetchedSessionBuffers)
             {
                 if (eventQueue.isEmpty())
                 {
                     allSessionBuffers.remove(buf);
                     removeSessionBuffer(buf);
                 }
                 else
                 {
                     unfetchedSessionBuffers.push(buf);
                 }
             }
         }

         private boolean waitForPromotion()
         {
             final Object promotionLock = this.promotionLock;

             long startTime = System.currentTimeMillis();
             long currentTime = System.currentTimeMillis();

             synchronized (promotionLock)
             {
                 while (this != leader && !shuttingDown)
                 {
                     // Calculate remaining keep-alive time
                     int keepAliveTime = getKeepAliveTime();
                     if (keepAliveTime > 0)
                     {
                         keepAliveTime -= (currentTime - startTime);
                     }
                     else
                     {
                         keepAliveTime = Integer.MAX_VALUE;
                     }

                     // Break the loop if there's no remaining keep-alive time.
                     if (keepAliveTime <= 0)
                     {
                         break;
                     }

                     // Wait for promotion
                     try
                     {
                         promotionLock.wait(keepAliveTime);
                     }
                     catch (InterruptedException e)
                     {
                     }

                     // Update currentTime for the next iteration
                     currentTime = System.currentTimeMillis();
                 }

                 boolean timeToLead = this == leader && !shuttingDown;

                 if (!timeToLead)
                 {
                     // time to die
                     synchronized (followers)
                     {
                         followers.remove(this);
                     }

                     // Mark as dead explicitly when we've got promotionLock.
                     dead = true;
                 }

                 return timeToLead;
             }
         }

         private void giveUpLead()
         {
             final Stack followers = ThreadPoolFilter.this.followers;
             Worker worker;
             do
             {
                 synchronized (followers)
                 {
                     worker = (Worker) followers.pop();
                 }

                 if (worker == null)
                 {
                     // Increase the number of threads if we
                     // are not shutting down and we can increase the number.
                     if (!shuttingDown
                             && getPoolSize() < getMaximumPoolSize())
                     {
                         worker = new Worker();
                         worker.lead();
                         worker.start();
                     }

                     // This loop should end because:
                     // 1) lead() is called already,
                     // 2) or it is shutting down and there's no more threads left.
                     break;
                 }
             }
             while (!worker.lead());
         }
     }

     protected static class EventType
     {
         public static final EventType OPENED = new EventType("OPENED");

         public static final EventType CLOSED = new EventType("CLOSED");

         public static final EventType READ = new EventType("READ");

         public static final EventType WRITTEN = new EventType("WRITTEN");

         public static final EventType RECEIVED = new EventType("RECEIVED");

         public static final EventType SENT = new EventType("SENT");

         public static final EventType IDLE = new EventType("IDLE");

         public static final EventType EXCEPTION = new EventType("EXCEPTION");

         private final String value;

         private EventType(String value)
         {
             this.value = value;
         }

         public String toString()
         {
             return value;
         }
     }

     protected static class Event
     {
         private final EventType type;
         private final NextFilter nextFilter;
         private final Object data;

         public Event(EventType type, NextFilter nextFilter, Object data)
         {
             this.type = type;
             this.nextFilter = nextFilter;
             this.data = data;
         }

         public Object getData()
         {
             return data;
         }


         public NextFilter getNextFilter()
         {
             return nextFilter;
         }


         public EventType getType()
         {
             return type;
         }
     }

     public void sessionCreated(NextFilter nextFilter, IoSession session)
     {
         nextFilter.sessionCreated(session);
     }

     public void sessionOpened(NextFilter nextFilter,
                               IoSession session)
     {
         fireEvent(nextFilter, session, EventType.OPENED, null);
     }

     public void sessionClosed(NextFilter nextFilter,
                               IoSession session)
     {
         fireEvent(nextFilter, session, EventType.CLOSED, null);
     }

     public void sessionIdle(NextFilter nextFilter,
                             IoSession session, IdleStatus status)
     {
         fireEvent(nextFilter, session, EventType.IDLE, status);
     }

     public void exceptionCaught(NextFilter nextFilter,
                                 IoSession session, Throwable cause)
     {
         fireEvent(nextFilter, session, EventType.EXCEPTION, cause);
     }

     public void messageReceived(NextFilter nextFilter,
                                 IoSession session, Object message)
     {
         ByteBufferUtil.acquireIfPossible(message);
         fireEvent(nextFilter, session, EventType.RECEIVED, message);
     }

     public void messageSent(NextFilter nextFilter,
                             IoSession session, Object message)
     {
         ByteBufferUtil.acquireIfPossible(message);
         fireEvent(nextFilter, session, EventType.SENT, message);
     }

     protected void processEvent(NextFilter nextFilter,
                                 IoSession session, EventType type,
                                 Object data)
     {
         if (type == EventType.RECEIVED)
         {
             nextFilter.messageReceived(session, data);
             ByteBufferUtil.releaseIfPossible(data);
         }
         else if (type == EventType.SENT)
         {
             nextFilter.messageSent(session, data);
             ByteBufferUtil.releaseIfPossible(data);
         }
         else if (type == EventType.EXCEPTION)
         {
             nextFilter.exceptionCaught(session, (Throwable) data);
         }
         else if (type == EventType.IDLE)
         {
             nextFilter.sessionIdle(session, (IdleStatus) data);
         }
         else if (type == EventType.OPENED)
         {
             nextFilter.sessionOpened(session);
         }
         else if (type == EventType.CLOSED)
         {
             nextFilter.sessionClosed(session);
         }
     }

     public void filterWrite(NextFilter nextFilter, IoSession session, WriteRequest writeRequest)
     {
         nextFilter.filterWrite(session, writeRequest);
     }

     public void filterClose(NextFilter nextFilter, IoSession session) throws Exception
     {
         nextFilter.filterClose(session);
     }
 }
	/*
	*
	* Licensed to the Apache Software Foundation (ASF) under one
	* or more contributor license agreements. See the NOTICE file
	* distributed with this work for additional information
	* regarding copyright ownership. The ASF licenses this file
	* to you under the Apache License, Version 2.0 (the
	* "License"); you may not use this file except in compliance
	* with the License. You may obtain a copy of the License at
	*
	* http://www.apache.org/licenses/LICENSE-2.0
	*
	* Unless required by applicable law or agreed to in writing,
	* software distributed under the License is distributed on an
	* "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
	* KIND, either express or implied. See the License for the
	* specific language governing permissions and limitations
	* under the License.
	*
	*/
	package org.apache.qpid.server.transport;

	import java.util.ArrayList;
	import java.util.IdentityHashMap;
	import java.util.Iterator;
	import java.util.List;
	import java.util.Map;
	import java.util.Set;

	import org.apache.mina.common.IdleStatus;
	import org.apache.mina.common.IoFilterAdapter;
	import org.apache.mina.common.IoHandler;
	import org.apache.mina.common.IoSession;
	import org.apache.mina.util.BlockingQueue;
	import org.apache.mina.util.ByteBufferUtil;
	import org.apache.mina.util.IdentityHashSet;
	import org.apache.mina.util.Queue;
	import org.apache.mina.util.Stack;

	/**
	* A Thread-pooling filter. This filter forwards {@link IoHandler} events
	* to its thread pool.
	* <p/>
	* This is an implementation of
	* <a href="http://deuce.doc.wustl.edu/doc/pspdfs/lf.pdf">Leader/Followers
	* thread pool</a> by Douglas C. Schmidt et al.
	*/
	public class ThreadPoolFilter extends IoFilterAdapter
	{
	/**
	* Default maximum size of thread pool (2G).
	*/
	public static final int DEFAULT_MAXIMUM_POOL_SIZE = Integer.MAX_VALUE;

	/**
	* Default keep-alive time of thread pool (1 min).
	*/
	public static final int DEFAULT_KEEP_ALIVE_TIME = 60 * 1000;

	/**
	* A queue which contains {@link Integer}s which represents reusable
	* thread IDs. {@link Worker} first checks this queue and then
	* uses {@link #threadId} when no reusable thread ID is available.
	*/
	private static final Queue threadIdReuseQueue = new Queue();
	private static int threadId = 0;

	private static int acquireThreadId()
	{
	synchronized (threadIdReuseQueue)
	{
	Integer id = (Integer) threadIdReuseQueue.pop();
	if (id == null)
	{
	return ++ threadId;
	}
	else
	{
	return id.intValue();
	}
	}
	}

	private static void releaseThreadId(int id)
	{
	synchronized (threadIdReuseQueue)
	{
	threadIdReuseQueue.push(new Integer(id));
	}
	}

	private final String threadNamePrefix;
	private final Map buffers = new IdentityHashMap();
	private final BlockingQueue unfetchedSessionBuffers = new BlockingQueue();
	private final Set allSessionBuffers = new IdentityHashSet();

	private Worker leader;
	private final Stack followers = new Stack();
	private final Set allWorkers = new IdentityHashSet();

	private int maximumPoolSize = DEFAULT_MAXIMUM_POOL_SIZE;
	private int keepAliveTime = DEFAULT_KEEP_ALIVE_TIME;

	private boolean shuttingDown;

	private int poolSize;
	private final Object poolSizeLock = new Object();

	/**
	* Creates a new instance of this filter with default thread pool settings.
	*/
	public ThreadPoolFilter()
	{
	this("IoThreadPool");
	}

	/**
	* Creates a new instance of this filter with the specified thread name prefix
	* and other default settings.
	*
	* @param threadNamePrefix the prefix of the thread names this pool will create.
	*/
	public ThreadPoolFilter(String threadNamePrefix)
	{
	if (threadNamePrefix == null)
	{
	throw new NullPointerException("threadNamePrefix");
	}
	threadNamePrefix = threadNamePrefix.trim();
	if (threadNamePrefix.length() == 0)
	{
	throw new IllegalArgumentException("threadNamePrefix is empty.");
	}
	this.threadNamePrefix = threadNamePrefix;
	}

	public String getThreadNamePrefix()
	{
	return threadNamePrefix;
	}

	public int getPoolSize()
	{
	synchronized (poolSizeLock)
	{
	return poolSize;
	}
	}

	public int getMaximumPoolSize()
	{
	return maximumPoolSize;
	}

	public int getKeepAliveTime()
	{
	return keepAliveTime;
	}

	public void setMaximumPoolSize(int maximumPoolSize)
	{
	if (maximumPoolSize <= 0)
	{
	throw new IllegalArgumentException();
	}
	this.maximumPoolSize = maximumPoolSize;
	}

	public void setKeepAliveTime(int keepAliveTime)
	{
	this.keepAliveTime = keepAliveTime;
	}

	public void init()
	{
	shuttingDown = false;
	leader = new Worker();
	leader.start();
	leader.lead();
	}

	public void destroy()
	{
	shuttingDown = true;
	int expectedPoolSize = 0;
	while (getPoolSize() != expectedPoolSize)
	{
	List allWorkers;
	synchronized (poolSizeLock)
	{
	allWorkers = new ArrayList(this.allWorkers);
	}

	// You may not interrupt the current thread.
	if (allWorkers.remove(Thread.currentThread()))
	{
	expectedPoolSize = 1;
	}

	for (Iterator i = allWorkers.iterator(); i.hasNext();)
	{
	Worker worker = (Worker) i.next();
	while (worker.isAlive())
	{
	worker.interrupt();
	try
	{
	// This timeout will help us from
	// infinite lock-up and interrupt workers again.
	worker.join(100);
	}
	catch (InterruptedException e)
	{
	}
	}
	}
	}

	this.allSessionBuffers.clear();
	this.unfetchedSessionBuffers.clear();
	this.buffers.clear();
	this.followers.clear();
	this.leader = null;
	}

	private void increasePoolSize(Worker worker)
	{
	synchronized (poolSizeLock)
	{
	poolSize++;
	allWorkers.add(worker);
	}
	}

	private void decreasePoolSize(Worker worker)
	{
	synchronized (poolSizeLock)
	{
	poolSize--;
	allWorkers.remove(worker);
	}
	}

	private void fireEvent(NextFilter nextFilter, IoSession session,
	EventType type, Object data)
	{
	final BlockingQueue unfetchedSessionBuffers = this.unfetchedSessionBuffers;
	final Set allSessionBuffers = this.allSessionBuffers;
	final Event event = new Event(type, nextFilter, data);

	synchronized (unfetchedSessionBuffers)
	{
	final SessionBuffer buf = getSessionBuffer(session);
	final Queue eventQueue = buf.eventQueue;

	synchronized (buf)
	{
	eventQueue.push(event);
	}

	if (!allSessionBuffers.contains(buf))
	{
	allSessionBuffers.add(buf);
	unfetchedSessionBuffers.push(buf);
	}
	}
	}

	/**
	* Implement this method to fetch (or pop) a {@link SessionBuffer} from
	* the given <tt>unfetchedSessionBuffers</tt>. The default implementation
	* simply pops the buffer from it. You could prioritize the fetch order.
	*
	* @return A non-null {@link SessionBuffer}
	*/
	protected SessionBuffer fetchSessionBuffer(Queue unfetchedSessionBuffers)
	{
	return (SessionBuffer) unfetchedSessionBuffers.pop();
	}

	private SessionBuffer getSessionBuffer(IoSession session)
	{
	final Map buffers = this.buffers;
	SessionBuffer buf = (SessionBuffer) buffers.get(session);
	if (buf == null)
	{
	synchronized (buffers)
	{
	buf = (SessionBuffer) buffers.get(session);
	if (buf == null)
	{
	buf = new SessionBuffer(session);
	buffers.put(session, buf);
	}
	}
	}
	return buf;
	}

	private void removeSessionBuffer(SessionBuffer buf)
	{
	final Map buffers = this.buffers;
	final IoSession session = buf.session;
	synchronized (buffers)
	{
	buffers.remove(session);
	}
	}

	protected static class SessionBuffer
	{
	private final IoSession session;

	private final Queue eventQueue = new Queue();

	private SessionBuffer(IoSession session)
	{
	this.session = session;
	}

	public IoSession getSession()
	{
	return session;
	}

	public Queue getEventQueue()
	{
	return eventQueue;
	}
	}

	private class Worker extends Thread
	{
	private final int id;
	private final Object promotionLock = new Object();
	private boolean dead;

	private Worker()
	{
	int id = acquireThreadId();
	this.id = id;
	this.setName(threadNamePrefix + '-' + id);
	increasePoolSize(this);
	}

	public boolean lead()
	{
	final Object promotionLock = this.promotionLock;
	synchronized (promotionLock)
	{
	if (dead)
	{
	return false;
	}

	leader = this;
	promotionLock.notify();
	}

	return true;
	}

	public void run()
	{
	for (; ;)
	{
	if (!waitForPromotion())
	{
	break;
	}

	SessionBuffer buf = fetchBuffer();
	giveUpLead();
	if (buf == null)
	{
	break;
	}

	processEvents(buf);
	follow();
	releaseBuffer(buf);
	}

	decreasePoolSize(this);
	releaseThreadId(id);
	}

	private SessionBuffer fetchBuffer()
	{
	BlockingQueue unfetchedSessionBuffers = ThreadPoolFilter.this.unfetchedSessionBuffers;
	synchronized (unfetchedSessionBuffers)
	{
	while (!shuttingDown)
	{
	try
	{
	unfetchedSessionBuffers.waitForNewItem();
	}
	catch (InterruptedException e)
	{
	continue;
	}

	return ThreadPoolFilter.this.fetchSessionBuffer(unfetchedSessionBuffers);
	}
	}

	return null;
	}

	private void processEvents(SessionBuffer buf)
	{
	final IoSession session = buf.session;
	final Queue eventQueue = buf.eventQueue;
	for (; ;)
	{
	Event event;
	synchronized (buf)
	{
	event = (Event) eventQueue.pop();
	if (event == null)
	{
	break;
	}
	}
	processEvent(event.getNextFilter(), session,
	event.getType(), event.getData());
	}
	}

	private void follow()
	{
	final Object promotionLock = this.promotionLock;
	final Stack followers = ThreadPoolFilter.this.followers;
	synchronized (promotionLock)
	{
	if (this != leader)
	{
	synchronized (followers)
	{
	followers.push(this);
	}
	}
	}
	}

	private void releaseBuffer(SessionBuffer buf)
	{
	final BlockingQueue unfetchedSessionBuffers = ThreadPoolFilter.this.unfetchedSessionBuffers;
	final Set allSessionBuffers = ThreadPoolFilter.this.allSessionBuffers;
	final Queue eventQueue = buf.eventQueue;

	synchronized (unfetchedSessionBuffers)
	{
	if (eventQueue.isEmpty())
	{
	allSessionBuffers.remove(buf);
	removeSessionBuffer(buf);
	}
	else
	{
	unfetchedSessionBuffers.push(buf);
	}
	}
	}

	private boolean waitForPromotion()
	{
	final Object promotionLock = this.promotionLock;

	long startTime = System.currentTimeMillis();
	long currentTime = System.currentTimeMillis();

	synchronized (promotionLock)
	{
	while (this != leader && !shuttingDown)
	{
	// Calculate remaining keep-alive time
	int keepAliveTime = getKeepAliveTime();
	if (keepAliveTime > 0)
	{
	keepAliveTime -= (currentTime - startTime);
	}
	else
	{
	keepAliveTime = Integer.MAX_VALUE;
	}

	// Break the loop if there's no remaining keep-alive time.
	if (keepAliveTime <= 0)
	{
	break;
	}

	// Wait for promotion
	try
	{
	promotionLock.wait(keepAliveTime);
	}
	catch (InterruptedException e)
	{
	}

	// Update currentTime for the next iteration
	currentTime = System.currentTimeMillis();
	}

	boolean timeToLead = this == leader && !shuttingDown;

	if (!timeToLead)
	{
	// time to die
	synchronized (followers)
	{
	followers.remove(this);
	}

	// Mark as dead explicitly when we've got promotionLock.
	dead = true;
	}

	return timeToLead;
	}
	}

	private void giveUpLead()
	{
	final Stack followers = ThreadPoolFilter.this.followers;
	Worker worker;
	do
	{
	synchronized (followers)
	{
	worker = (Worker) followers.pop();
	}

	if (worker == null)
	{
	// Increase the number of threads if we
	// are not shutting down and we can increase the number.
	if (!shuttingDown
	&& getPoolSize() < getMaximumPoolSize())
	{
	worker = new Worker();
	worker.lead();
	worker.start();
	}

	// This loop should end because:
	// 1) lead() is called already,
	// 2) or it is shutting down and there's no more threads left.
	break;
	}
	}
	while (!worker.lead());
	}
	}

	protected static class EventType
	{
	public static final EventType OPENED = new EventType("OPENED");

	public static final EventType CLOSED = new EventType("CLOSED");

	public static final EventType READ = new EventType("READ");

	public static final EventType WRITTEN = new EventType("WRITTEN");

	public static final EventType RECEIVED = new EventType("RECEIVED");

	public static final EventType SENT = new EventType("SENT");

	public static final EventType IDLE = new EventType("IDLE");

	public static final EventType EXCEPTION = new EventType("EXCEPTION");

	private final String value;

	private EventType(String value)
	{
	this.value = value;
	}

	public String toString()
	{
	return value;
	}
	}

	protected static class Event
	{
	private final EventType type;
	private final NextFilter nextFilter;
	private final Object data;

	public Event(EventType type, NextFilter nextFilter, Object data)
	{
	this.type = type;
	this.nextFilter = nextFilter;
	this.data = data;
	}

	public Object getData()
	{
	return data;
	}


	public NextFilter getNextFilter()
	{
	return nextFilter;
	}


	public EventType getType()
	{
	return type;
	}
	}

	public void sessionCreated(NextFilter nextFilter, IoSession session)
	{
	nextFilter.sessionCreated(session);
	}

	public void sessionOpened(NextFilter nextFilter,
	IoSession session)
	{
	fireEvent(nextFilter, session, EventType.OPENED, null);
	}

	public void sessionClosed(NextFilter nextFilter,
	IoSession session)
	{
	fireEvent(nextFilter, session, EventType.CLOSED, null);
	}

	public void sessionIdle(NextFilter nextFilter,
	IoSession session, IdleStatus status)
	{
	fireEvent(nextFilter, session, EventType.IDLE, status);
	}

	public void exceptionCaught(NextFilter nextFilter,
	IoSession session, Throwable cause)
	{
	fireEvent(nextFilter, session, EventType.EXCEPTION, cause);
	}

	public void messageReceived(NextFilter nextFilter,
	IoSession session, Object message)
	{
	ByteBufferUtil.acquireIfPossible(message);
	fireEvent(nextFilter, session, EventType.RECEIVED, message);
	}

	public void messageSent(NextFilter nextFilter,
	IoSession session, Object message)
	{
	ByteBufferUtil.acquireIfPossible(message);
	fireEvent(nextFilter, session, EventType.SENT, message);
	}

	protected void processEvent(NextFilter nextFilter,
	IoSession session, EventType type,
	Object data)
	{
	if (type == EventType.RECEIVED)
	{
	nextFilter.messageReceived(session, data);
	ByteBufferUtil.releaseIfPossible(data);
	}
	else if (type == EventType.SENT)
	{
	nextFilter.messageSent(session, data);
	ByteBufferUtil.releaseIfPossible(data);
	}
	else if (type == EventType.EXCEPTION)
	{
	nextFilter.exceptionCaught(session, (Throwable) data);
	}
	else if (type == EventType.IDLE)
	{
	nextFilter.sessionIdle(session, (IdleStatus) data);
	}
	else if (type == EventType.OPENED)
	{
	nextFilter.sessionOpened(session);
	}
	else if (type == EventType.CLOSED)
	{
	nextFilter.sessionClosed(session);
	}
	}

	public void filterWrite(NextFilter nextFilter, IoSession session, WriteRequest writeRequest)
	{
	nextFilter.filterWrite(session, writeRequest);
	}

	public void filterClose(NextFilter nextFilter, IoSession session) throws Exception
	{
	nextFilter.filterClose(session);
	}
	}