DefaultThreadPool


1   /*
2    * Copyright 1999-2004 The Apache Software Foundation.
3    *
4    * Licensed under the Apache License, Version 2.0 (the "License");
5    * you may not use this file except in compliance with the License.
6    * You may obtain a copy of the License at
7    *
8    *      http://www.apache.org/licenses/LICENSE-2.0
9    *
10   * Unless required by applicable law or agreed to in writing, software
11   * distributed under the License is distributed on an "AS IS" BASIS,
12   * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13   * See the License for the specific language governing permissions and
14   * limitations under the License.
15   */
16  package org.apache.cocoon.components.thread;
17  
18  import org.apache.avalon.framework.logger.LogEnabled;
19  import org.apache.avalon.framework.logger.Logger;
20  
21  import EDU.oswego.cs.dl.util.concurrent.PooledExecutor;
22  
23  
24  /**
25   * The DefaultThreadPool class implements the {@link ThreadPool} interface.
26   * Instances of this class are made by the {@link RunnableManager} passing a
27   * configuration into the <code>configure</code> method.
28   *
29   * @author <a HREF="mailto:giacomo.at.apache.org">Giacomo Pati</a>
30   * @version CVS $Id: DefaultThreadPool.java 56843 2004-11-07 13:34:30Z giacomo $
31   */
32  public class DefaultThreadPool
33      extends PooledExecutor
34      implements ThreadPool, LogEnabled
35  {
36      //~ Static fields/initializers ---------------------------------------------
37  
38      /** Default ThreadPool block policy */
39      public static final String   POLICY_DEFAULT = POLICY_RUN;
40  
41      //~ Instance fields --------------------------------------------------------
42  
43      /** Wrapps a channel */
44      private ChannelWrapper m_channelWrapper;
45  
46      /** Our logger */
47      private Logger m_logger;
48  
49      /** The Queue */
50      private Queue m_queue;
51  
52      /** The blocking policy */
53      private String   m_blockPolicy;
54  
55      /** The name of this thread pool */
56      private String   m_name;
57  
58      /** Should we wait for running jobs to terminate on shutdown ? */
59      private boolean m_shutdownGraceful;
60  
61      /** The maximum queue size */
62      private int m_queueSize;
63  
64      /** How long to wait for running jobs to terminate on disposition */
65      private int m_shutdownWaitTimeMs;
66  
67      //~ Constructors -----------------------------------------------------------
68  
69      /**
70       * Create a new pool.
71       */
72      DefaultThreadPool(  )
73      {
74          this( new ChannelWrapper(  ) );
75      }
76  
77      /**
78       * Create a new pool.
79       *
80       * @param channel DOCUMENT ME!
81       */
82      private DefaultThreadPool( final ChannelWrapper channel )
83      {
84          super( channel );
85          m_channelWrapper = channel;
86      }
87  
88      //~ Methods ----------------------------------------------------------------
89  
90      /**
91       * DOCUMENT ME!
92       *
93       * @return Returns the blockPolicy.
94       */
95      public String   getBlockPolicy(  )
96      {
97          return m_blockPolicy;
98      }
99  
100     /**
101      * DOCUMENT ME!
102      *
103      * @return maximum size of the queue (0 if isQueued() == false)
104      *
105      * @see org.apache.cocoon.components.thread.ThreadPool#getQueueSize()
106      */
107     public int getMaxQueueSize(  )
108     {
109         return ( ( m_queueSize < 0 ) ? Integer.MAX_VALUE : m_queueSize );
110     }
111 
112     /**
113      * DOCUMENT ME!
114      *
115      * @return size of queue (0 if isQueued() == false)
116      *
117      * @see org.apache.cocoon.components.thread.ThreadPool#getQueueSize()
118      */
119     public int getMaximumQueueSize(  )
120     {
121         return m_queueSize;
122     }
123 
124     /**
125      * @see org.apache.cocoon.components.thread.ThreadPool#getName()
126      */
127     public String   getName(  )
128     {
129         return m_name;
130     }
131 
132     /**
133      * Get hte priority used to create Threads
134      *
135      * @return {@link Thread#MIN_PRIORITY}, {@link Thread#NORM_PRIORITY}, or
136      *         {@link Thread#MAX_PRIORITY}
137      */
138     public int getPriority(  )
139     {
140         return ((ThreadFactory)super.getThreadFactory()).getPriority();
141     }
142 
143     /**
144      * DOCUMENT ME!
145      *
146      * @return current size of the queue (0 if isQueued() == false)
147      *
148      * @see org.apache.cocoon.components.thread.ThreadPool#getQueueSize()
149      */
150     public int getQueueSize(  )
151     {
152         return m_queue.getQueueSize(  );
153     }
154 
155     /**
156      * Whether this DefaultThreadPool has a queue
157      *
158      * @return Returns the m_isQueued.
159      *
160      * @see org.apache.cocoon.components.thread.ThreadPool#isQueued()
161      */
162     public boolean isQueued(  )
163     {
164         return m_queueSize != 0;
165     }
166 
167     /**
168      * Set the logger
169      *
170      * @param logger
171      *
172      * @see org.apache.avalon.framework.logger.LogEnabled#enableLogging(org.apache.avalon.framework.logger.Logger)
173      */
174     public void enableLogging( Logger logger )
175     {
176         m_logger = logger;
177     }
178 
179     /**
180      * Execute a command
181      *
182      * @param command The {@link Runnable} to execute
183      *
184      * @throws InterruptedException In case of interruption
185      */
186     public void execute( Runnable   command )
187         throws InterruptedException  
188     {
189         if( getLogger(  ).isDebugEnabled(  ) )
190         {
191             getLogger(  ).debug( "Executing Command: " + command.toString(  ) +
192                                  ",pool=" + getName(  ) );
193         }
194 
195         super.execute( command );
196     }
197 
198     /**
199      * @see org.apache.cocoon.components.thread.ThreadPool#shutdown()
200      */
201     public void shutdown(  )
202     {
203         if( m_shutdownGraceful )
204         {
205             shutdownAfterProcessingCurrentlyQueuedTasks(  );
206         }
207         else
208         {
209             shutdownNow(  );
210         }
211 
212         try
213         {
214             if( getShutdownWaitTimeMs(  ) > 0 )
215             {
216                 if( ! awaitTerminationAfterShutdown( getShutdownWaitTimeMs(  ) ) )
217                 {
218                     getLogger(  ).warn( "running commands have not terminated within " +
219                                         getShutdownWaitTimeMs(  ) +
220                                         "ms. Will shut them down by interruption" );
221                     interruptAll(  );
222                     shutdownNow(  );
223                 }
224             }
225 
226             awaitTerminationAfterShutdown(  );
227         }
228         catch( final InterruptedException   ie )
229         {
230             getLogger(  ).error( "cannot shutdown ThreadPool", ie );
231         }
232     }
233 
234     /**
235      * Set the blocking policy
236      *
237      * @param blockPolicy The blocking policy value
238      */
239     void setBlockPolicy( final String   blockPolicy )
240     {
241         m_blockPolicy = blockPolicy;
242 
243         if( POLICY_ABORT.equalsIgnoreCase( blockPolicy ) )
244         {
245             abortWhenBlocked(  );
246         }
247         else if( POLICY_DISCARD.equalsIgnoreCase( blockPolicy ) )
248         {
249             discardWhenBlocked(  );
250         }
251         else if( POLICY_DISCARD_OLDEST.equalsIgnoreCase( blockPolicy ) )
252         {
253             discardOldestWhenBlocked(  );
254         }
255         else if( POLICY_RUN.equalsIgnoreCase( blockPolicy ) )
256         {
257             runWhenBlocked(  );
258         }
259         else if( POLICY_WAIT.equalsIgnoreCase( blockPolicy ) )
260         {
261             waitWhenBlocked(  );
262         }
263         else
264         {
265             final StringBuffer   msg = new StringBuffer  (  );
266             msg.append( "WARNING: Unknown block-policy configuration \"" )
267                .append( blockPolicy );
268             msg.append( "\". Should be one of \"" ).append( POLICY_ABORT );
269             msg.append( "\",\"" ).append( POLICY_DISCARD );
270             msg.append( "\",\"" ).append( POLICY_DISCARD_OLDEST );
271             msg.append( "\",\"" ).append( POLICY_RUN );
272             msg.append( "\",\"" ).append( POLICY_WAIT );
273             msg.append( "\". Will use \"" ).append( POLICY_DEFAULT ).append( "\"" );
274             getLogger(  ).warn( msg.toString(  ) );
275             setBlockPolicy( POLICY_DEFAULT );
276         }
277     }
278 
279     /**
280      * DOCUMENT ME!
281      *
282      * @param name The name to set.
283      */
284     void setName( String   name )
285     {
286         m_name = name;
287     }
288 
289     /**
290      * DOCUMENT ME!
291      *
292      * @param queueSize DOCUMENT ME!
293      */
294     void setQueue( final int queueSize )
295     {
296         if( queueSize != 0 )
297         {
298             if( queueSize > 0 )
299             {
300                 m_queue = new BoundedQueue( queueSize );
301             }
302             else
303             {
304                 m_queue = new LinkedQueue(  );
305             }
306         }
307         else
308         {
309             m_queue = new SynchronousChannel(  );
310         }
311 
312         m_queueSize = queueSize;
313         m_channelWrapper.setChannel( m_queue );
314     }
315 
316     /**
317      * DOCUMENT ME!
318      *
319      * @param shutdownGraceful The shutdownGraceful to set.
320      */
321     void setShutdownGraceful( boolean shutdownGraceful )
322     {
323         m_shutdownGraceful = shutdownGraceful;
324     }
325 
326     /**
327      * DOCUMENT ME!
328      *
329      * @return Returns the shutdownGraceful.
330      */
331     boolean isShutdownGraceful(  )
332     {
333         return m_shutdownGraceful;
334     }
335 
336     /**
337      * DOCUMENT ME!
338      *
339      * @param shutdownWaitTimeMs The shutdownWaitTimeMs to set.
340      */
341     void setShutdownWaitTimeMs( int shutdownWaitTimeMs )
342     {
343         m_shutdownWaitTimeMs = shutdownWaitTimeMs;
344     }
345 
346     /**
347      * DOCUMENT ME!
348      *
349      * @return Returns the shutdownWaitTimeMs.
350      */
351     int getShutdownWaitTimeMs(  )
352     {
353         return m_shutdownWaitTimeMs;
354     }
355 
356     /**
357      * Get our <code>Logger</code>
358      *
359      * @return our <code>Logger</code>
360      */
361     private Logger getLogger(  )
362     {
363         return m_logger;
364     }
365 }
366
A to Z: JavaDoc & Examples Daily Java News & Articles Open Source Projects Open Source Codes Free Computer Books Remove Frame
Popular Tags