1 /*
2 * Copyright (c) 2003, 2004, Oracle and/or its affiliates. All rights reserved.
3 * DO NOT ALTER OR REMOVE COPYRIGHT NOTICES OR THIS FILE HEADER.
4 *
5 * This code is free software; you can redistribute it and/or modify it
6 * under the terms of the GNU General Public License version 2 only, as
7 * published by the Free Software Foundation. Oracle designates this
8 * particular file as subject to the "Classpath" exception as provided
9 * by Oracle in the LICENSE file that accompanied this code.
10 *
11 * This code is distributed in the hope that it will be useful, but WITHOUT
12 * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
13 * FITNESS FOR A PARTICULAR PURPOSE. See the GNU General Public License
14 * version 2 for more details (a copy is included in the LICENSE file that
15 * accompanied this code).
16 *
17 * You should have received a copy of the GNU General Public License version
18 * 2 along with this work; if not, write to the Free Software Foundation,
19 * Inc., 51 Franklin St, Fifth Floor, Boston, MA 02110-1301 USA.
20 *
21 * Please contact Oracle, 500 Oracle Parkway, Redwood Shores, CA 94065 USA
22 * or visit www.oracle.com if you need additional information or have any
23 * questions.
24 */
25
26 package com.sun.corba.se.impl.orbutil.threadpool;
27
28 import com.sun.corba.se.spi.orbutil.threadpool.NoSuchWorkQueueException;
29 import com.sun.corba.se.spi.orbutil.threadpool.ThreadPool;
30 import com.sun.corba.se.spi.orbutil.threadpool.Work;
31 import com.sun.corba.se.spi.orbutil.threadpool.WorkQueue;
32
33 import com.sun.corba.se.impl.orbutil.ORBConstants;
34 import com.sun.corba.se.impl.orbutil.threadpool.WorkQueueImpl;
35
36 import com.sun.corba.se.spi.monitoring.MonitoringConstants;
37 import com.sun.corba.se.spi.monitoring.MonitoredObject;
38 import com.sun.corba.se.spi.monitoring.MonitoringFactories;
39 import com.sun.corba.se.spi.monitoring.LongMonitoredAttributeBase;
40
41 public class ThreadPoolImpl implements ThreadPool
42 {
43 private static int threadCounter = 0; // serial counter useful for debugging
44
45 private WorkQueue workQueue;
46
47 // Stores the number of available worker threads
48 private int availableWorkerThreads = 0;
49
50 // Stores the number of threads in the threadpool currently
51 private int currentThreadCount = 0;
52
53 // Minimum number of worker threads created at instantiation of the threadpool
54 private int minWorkerThreads = 0;
55
56 // Maximum number of worker threads in the threadpool
57 private int maxWorkerThreads = 0;
58
59 // Inactivity timeout value for worker threads to exit and stop running
60 private long inactivityTimeout;
61
62 // Indicates if the threadpool is bounded or unbounded
63 private boolean boundedThreadPool = false;
64
65 // Running count of the work items processed
66 // Set the value to 1 so that divide by zero is avoided in
67 // averageWorkCompletionTime()
68 private long processedCount = 1;
69
70 // Running aggregate of the time taken in millis to execute work items
71 // processed by the threads in the threadpool
72 private long totalTimeTaken = 0;
73
74 // Lock for protecting state when required
75 private Object lock = new Object();
76
77 // Name of the ThreadPool
78 private String name;
79
80 // MonitoredObject for ThreadPool
81 private MonitoredObject threadpoolMonitoredObject;
82
83 // ThreadGroup in which threads should be created
84 private ThreadGroup threadGroup ;
85
86 /**
87 * This constructor is used to create an unbounded threadpool
88 */
89 public ThreadPoolImpl(ThreadGroup tg, String threadpoolName) {
90 inactivityTimeout = ORBConstants.DEFAULT_INACTIVITY_TIMEOUT;
91 maxWorkerThreads = Integer.MAX_VALUE;
92 workQueue = new WorkQueueImpl(this);
93 threadGroup = tg ;
94 name = threadpoolName;
95 initializeMonitoring();
96 }
97
98 /**
99 * This constructor is used to create an unbounded threadpool
100 * in the ThreadGroup of the current thread
101 */
102 public ThreadPoolImpl(String threadpoolName) {
103 this( Thread.currentThread().getThreadGroup(), threadpoolName ) ;
104 }
105
106 /**
107 * This constructor is used to create bounded threadpool
108 */
109 public ThreadPoolImpl(int minSize, int maxSize, long timeout,
110 String threadpoolName)
111 {
112 minWorkerThreads = minSize;
113 maxWorkerThreads = maxSize;
114 inactivityTimeout = timeout;
115 boundedThreadPool = true;
116 workQueue = new WorkQueueImpl(this);
117 name = threadpoolName;
118 for (int i = 0; i < minWorkerThreads; i++) {
119 createWorkerThread();
120 }
121 initializeMonitoring();
122 }
123
124 // Setup monitoring for this threadpool
125 private void initializeMonitoring() {
126 // Get root monitored object
127 MonitoredObject root = MonitoringFactories.getMonitoringManagerFactory().
128 createMonitoringManager(MonitoringConstants.DEFAULT_MONITORING_ROOT, null).
129 getRootMonitoredObject();
130
131 // Create the threadpool monitoring root
132 MonitoredObject threadPoolMonitoringObjectRoot = root.getChild(
133 MonitoringConstants.THREADPOOL_MONITORING_ROOT);
134 if (threadPoolMonitoringObjectRoot == null) {
135 threadPoolMonitoringObjectRoot = MonitoringFactories.
136 getMonitoredObjectFactory().createMonitoredObject(
137 MonitoringConstants.THREADPOOL_MONITORING_ROOT,
138 MonitoringConstants.THREADPOOL_MONITORING_ROOT_DESCRIPTION);
139 root.addChild(threadPoolMonitoringObjectRoot);
140 }
141 threadpoolMonitoredObject = MonitoringFactories.
142 getMonitoredObjectFactory().
143 createMonitoredObject(name,
200
201 public WorkQueue getAnyWorkQueue()
202 {
203 return workQueue;
204 }
205
206 public WorkQueue getWorkQueue(int queueId)
207 throws NoSuchWorkQueueException
208 {
209 if (queueId != 0)
210 throw new NoSuchWorkQueueException();
211 return workQueue;
212 }
213
214 /**
215 * To be called from the workqueue when work is added to the
216 * workQueue. This method would create new threads if required
217 * or notify waiting threads on the queue for available work
218 */
219 void notifyForAvailableWork(WorkQueue aWorkQueue) {
220 synchronized (lock) {
221 if (availableWorkerThreads == 0) {
222 createWorkerThread();
223 } else {
224 aWorkQueue.notify();
225 }
226 }
227 }
228
229
230 /**
231 * To be called from the workqueue to create worker threads when none
232 * available.
233 */
234 void createWorkerThread() {
235 WorkerThread thread;
236
237 synchronized (lock) {
238 if (boundedThreadPool) {
239 if (currentThreadCount < maxWorkerThreads) {
240 thread = new WorkerThread(threadGroup, getName());
241 currentThreadCount++;
242 } else {
243 // REVIST - Need to create a thread to monitor the
244 // the state for deadlock i.e. all threads waiting for
245 // something which can be got from the item in the
246 // workqueue, but there is no thread available to
247 // process that work item - DEADLOCK !!
248 return;
249 }
250 } else {
251 thread = new WorkerThread(threadGroup, getName());
252 currentThreadCount++;
253 }
254 }
255
256 // The thread must be set to a daemon thread so the
257 // VM can exit if the only threads left are PooledThreads
258 // or other daemons. We don't want to rely on the
259 // calling thread always being a daemon.
260
261 // Catch exceptions since setDaemon can cause a
262 // security exception to be thrown under netscape
263 // in the Applet mode
264 try {
265 thread.setDaemon(true);
266 } catch (Exception e) {
267 // REVISIT - need to do some logging here
268 }
269
270 thread.start();
271 }
272
273 /**
274 * This method will return the minimum number of threads maintained
275 * by the threadpool.
276 */
277 public int minimumNumberOfThreads() {
278 return minWorkerThreads;
279 }
280
281 /**
282 * This method will return the maximum number of threads in the
283 * threadpool at any point in time, for the life of the threadpool
284 */
285 public int maximumNumberOfThreads() {
286 return maxWorkerThreads;
287 }
288
289 /**
290 * This method will return the time in milliseconds when idle
291 * threads in the threadpool are removed.
292 */
293 public long idleTimeoutForThreads() {
294 return inactivityTimeout;
295 }
296
297 /**
298 * This method will return the total number of threads currently in the
299 * threadpool. This method returns a value which is not synchronized.
300 */
301 public int currentNumberOfThreads() {
302 synchronized (lock) {
303 return currentThreadCount;
304 }
305 }
306
307 /**
308 * This method will return the number of available threads in the
309 * threadpool which are waiting for work. This method returns a
310 * value which is not synchronized.
311 */
312 public int numberOfAvailableThreads() {
313 synchronized (lock) {
314 return availableWorkerThreads;
315 }
316 }
317
318 /**
319 * This method will return the number of busy threads in the threadpool
320 * This method returns a value which is not synchronized.
321 */
322 public int numberOfBusyThreads() {
323 synchronized (lock) {
324 return (currentThreadCount - availableWorkerThreads);
325 }
326 }
327
328 /**
329 * This method returns the average elapsed time taken to complete a Work
330 * item in milliseconds.
331 */
332 public long averageWorkCompletionTime() {
333 synchronized (lock) {
334 return (totalTimeTaken / processedCount);
335 }
336 }
337
338 /**
339 * This method returns the number of Work items processed by the threadpool
340 */
341 public long currentProcessedCount() {
342 synchronized (lock) {
343 return processedCount;
344 }
345 }
346
347 public String getName() {
348 return name;
349 }
350
351 /**
352 * This method will return the number of WorkQueues serviced by the threadpool.
353 */
354 public int numberOfWorkQueues() {
355 return 1;
356 }
357
358
359 private static synchronized int getUniqueThreadId() {
360 return ThreadPoolImpl.threadCounter++;
361 }
362
363
364 private class WorkerThread extends Thread
365 {
366 private Work currentWork;
367 private int threadId = 0; // unique id for the thread
368 // thread pool this WorkerThread belongs too
369 private String threadPoolName;
370 // name seen by Thread.getName()
371 private StringBuffer workerThreadName = new StringBuffer();
372
373 WorkerThread(ThreadGroup tg, String threadPoolName) {
374 super(tg, "Idle");
375 this.threadId = ThreadPoolImpl.getUniqueThreadId();
376 this.threadPoolName = threadPoolName;
377 setName(composeWorkerThreadName(threadPoolName, "Idle"));
378 }
379
380 public void run() {
381 while (true) {
382 try {
383
384 synchronized (lock) {
385 availableWorkerThreads++;
386 }
387
388 // Get some work to do
389 currentWork = ((WorkQueueImpl)workQueue).requestWork(inactivityTimeout);
390
391 synchronized (lock) {
392 availableWorkerThreads--;
393 // It is possible in notifyForAvailableWork that the
394 // check for availableWorkerThreads = 0 may return
395 // false, because the availableWorkerThreads has not been
396 // decremented to zero before the producer thread added
397 // work to the queue. This may create a deadlock, if the
398 // executing thread needs information which is in the work
399 // item queued in the workqueue, but has no thread to work
400 // on it since none was created because availableWorkerThreads = 0
401 // returned false.
402 // The following code will ensure that a thread is always available
403 // in those situations
404 if ((availableWorkerThreads == 0) &&
405 (workQueue.workItemsInQueue() > 0)) {
406 createWorkerThread();
407 }
408 }
409
410 // Set the thread name for debugging.
411 setName(composeWorkerThreadName(threadPoolName,
412 Integer.toString(this.threadId)));
413
414 long start = System.currentTimeMillis();
415
416 try {
417 // Do the work
418 currentWork.doWork();
419 } catch (Throwable t) {
420 // Ignore all errors.
421 ;
422 }
423
424 long end = System.currentTimeMillis();
425
426
427 synchronized (lock) {
428 totalTimeTaken += (end - start);
429 processedCount++;
430 }
431
432 // set currentWork to null so that the work item can be
433 // garbage collected
434 currentWork = null;
435
436 setName(composeWorkerThreadName(threadPoolName, "Idle"));
437
438 } catch (TimeoutException e) {
439 // This thread timed out waiting for something to do.
440
441 synchronized (lock) {
442 availableWorkerThreads--;
443
444 // This should for both bounded and unbounded case
445 if (currentThreadCount > minWorkerThreads) {
446 currentThreadCount--;
447 // This thread can exit.
448 return;
449 } else {
450 // Go back to waiting on workQueue
451 continue;
452 }
453 }
454 } catch (InterruptedException ie) {
455 // InterruptedExceptions are
456 // caught here. Thus, threads can be forced out of
457 // requestWork and so they have to reacquire the lock.
458 // Other options include ignoring or
459 // letting this thread die.
460 // Ignoring for now. REVISIT
461 synchronized (lock) {
462 availableWorkerThreads--;
463 }
464
465 } catch (Throwable e) {
466
467 // Ignore any exceptions that currentWork.process
468 // accidently lets through, but let Errors pass.
469 // Add debugging output? REVISIT
470 synchronized (lock) {
471 availableWorkerThreads--;
472 }
473
474 }
475 }
476 }
477
478 private String composeWorkerThreadName(String poolName, String workerName) {
479 workerThreadName.setLength(0);
480 workerThreadName.append("p: ").append(poolName);
481 workerThreadName.append("; w: ").append(workerName);
482 return workerThreadName.toString();
483 }
484 } // End of WorkerThread class
485
486 }
487
488 // End of file.
|
1 /*
2 * Copyright (c) 2003, 2012, Oracle and/or its affiliates. All rights reserved.
3 * DO NOT ALTER OR REMOVE COPYRIGHT NOTICES OR THIS FILE HEADER.
4 *
5 * This code is free software; you can redistribute it and/or modify it
6 * under the terms of the GNU General Public License version 2 only, as
7 * published by the Free Software Foundation. Oracle designates this
8 * particular file as subject to the "Classpath" exception as provided
9 * by Oracle in the LICENSE file that accompanied this code.
10 *
11 * This code is distributed in the hope that it will be useful, but WITHOUT
12 * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
13 * FITNESS FOR A PARTICULAR PURPOSE. See the GNU General Public License
14 * version 2 for more details (a copy is included in the LICENSE file that
15 * accompanied this code).
16 *
17 * You should have received a copy of the GNU General Public License version
18 * 2 along with this work; if not, write to the Free Software Foundation,
19 * Inc., 51 Franklin St, Fifth Floor, Boston, MA 02110-1301 USA.
20 *
21 * Please contact Oracle, 500 Oracle Parkway, Redwood Shores, CA 94065 USA
22 * or visit www.oracle.com if you need additional information or have any
23 * questions.
24 */
25
26 package com.sun.corba.se.impl.orbutil.threadpool;
27
28 import java.io.IOException;
29 import java.io.Closeable;
30
31 import java.security.AccessController;
32 import java.security.PrivilegedAction;
33
34 import java.util.List;
35 import java.util.ArrayList;
36
37 import java.util.concurrent.atomic.AtomicInteger;
38 import java.util.concurrent.atomic.AtomicLong;
39
40 import com.sun.corba.se.spi.orbutil.threadpool.NoSuchWorkQueueException;
41 import com.sun.corba.se.spi.orbutil.threadpool.ThreadPool;
42 import com.sun.corba.se.spi.orbutil.threadpool.Work;
43 import com.sun.corba.se.spi.orbutil.threadpool.WorkQueue;
44
45 import com.sun.corba.se.impl.orbutil.ORBConstants;
46 import com.sun.corba.se.impl.orbutil.threadpool.WorkQueueImpl;
47
48 import com.sun.corba.se.spi.monitoring.MonitoringConstants;
49 import com.sun.corba.se.spi.monitoring.MonitoredObject;
50 import com.sun.corba.se.spi.monitoring.MonitoringFactories;
51 import com.sun.corba.se.spi.orb.ORB;
52 import com.sun.corba.se.spi.monitoring.LongMonitoredAttributeBase;
53
54 import com.sun.corba.se.impl.logging.ORBUtilSystemException;
55 import com.sun.corba.se.impl.orbutil.ORBConstants;
56 import com.sun.corba.se.spi.logging.CORBALogDomains;
57
58 public class ThreadPoolImpl implements ThreadPool
59 {
60 // serial counter useful for debugging
61 private static AtomicInteger threadCounter = new AtomicInteger(0);
62 private static final ORBUtilSystemException wrapper =
63 ORBUtilSystemException.get(CORBALogDomains.RPC_TRANSPORT);
64
65
66 // Any time currentThreadCount and/or availableWorkerThreads is updated
67 // or accessed this ThreadPool's WorkQueue must be locked. And, it is
68 // expected that this ThreadPool's WorkQueue is the only object that
69 // updates and accesses these values directly and indirectly though a
70 // call to a method in this ThreadPool. If any call to update or access
71 // those values must synchronized on this ThreadPool's WorkQueue.
72 private WorkQueue workQueue;
73
74 // Stores the number of available worker threads
75 private int availableWorkerThreads = 0;
76
77 // Stores the number of threads in the threadpool currently
78 private int currentThreadCount = 0;
79
80 // Minimum number of worker threads created at instantiation of the threadpool
81 private int minWorkerThreads = 0;
82
83 // Maximum number of worker threads in the threadpool
84 private int maxWorkerThreads = 0;
85
86 // Inactivity timeout value for worker threads to exit and stop running
87 private long inactivityTimeout;
88
89 // Indicates if the threadpool is bounded or unbounded
90 private boolean boundedThreadPool = false;
91
92 // Running count of the work items processed
93 // Set the value to 1 so that divide by zero is avoided in
94 // averageWorkCompletionTime()
95 private AtomicLong processedCount = new AtomicLong(1);
96
97 // Running aggregate of the time taken in millis to execute work items
98 // processed by the threads in the threadpool
99 private AtomicLong totalTimeTaken = new AtomicLong(0);
100
101 // Name of the ThreadPool
102 private String name;
103
104 // MonitoredObject for ThreadPool
105 private MonitoredObject threadpoolMonitoredObject;
106
107 // ThreadGroup in which threads should be created
108 private ThreadGroup threadGroup;
109
110 Object workersLock = new Object();
111 List<WorkerThread> workers = new ArrayList<>();
112
113 /**
114 * This constructor is used to create an unbounded threadpool
115 */
116 public ThreadPoolImpl(ThreadGroup tg, String threadpoolName) {
117 inactivityTimeout = ORBConstants.DEFAULT_INACTIVITY_TIMEOUT;
118 maxWorkerThreads = Integer.MAX_VALUE;
119 workQueue = new WorkQueueImpl(this);
120 threadGroup = tg;
121 name = threadpoolName;
122 initializeMonitoring();
123 }
124
125 /**
126 * This constructor is used to create an unbounded threadpool
127 * in the ThreadGroup of the current thread
128 */
129 public ThreadPoolImpl(String threadpoolName) {
130 this( Thread.currentThread().getThreadGroup(), threadpoolName ) ;
131 }
132
133 /**
134 * This constructor is used to create bounded threadpool
135 */
136 public ThreadPoolImpl(int minSize, int maxSize, long timeout,
137 String threadpoolName)
138 {
139 minWorkerThreads = minSize;
140 maxWorkerThreads = maxSize;
141 inactivityTimeout = timeout;
142 boundedThreadPool = true;
143 workQueue = new WorkQueueImpl(this);
144 name = threadpoolName;
145 for (int i = 0; i < minWorkerThreads; i++) {
146 createWorkerThread();
147 }
148 initializeMonitoring();
149 }
150
151 // Note that this method should not return until AFTER all threads have died.
152 public void close() throws IOException {
153
154 // Copy to avoid concurrent modification problems.
155 List<WorkerThread> copy = null;
156 synchronized (workersLock) {
157 copy = new ArrayList<>(workers);
158 }
159
160 for (WorkerThread wt : copy) {
161 wt.close();
162 while (wt.getState() != Thread.State.TERMINATED) {
163 try {
164 wt.join();
165 } catch (InterruptedException exc) {
166 wrapper.interruptedJoinCallWhileClosingThreadPool(exc, wt, this);
167 }
168 }
169 }
170
171 threadGroup = null;
172 }
173
174
175 // Setup monitoring for this threadpool
176 private void initializeMonitoring() {
177 // Get root monitored object
178 MonitoredObject root = MonitoringFactories.getMonitoringManagerFactory().
179 createMonitoringManager(MonitoringConstants.DEFAULT_MONITORING_ROOT, null).
180 getRootMonitoredObject();
181
182 // Create the threadpool monitoring root
183 MonitoredObject threadPoolMonitoringObjectRoot = root.getChild(
184 MonitoringConstants.THREADPOOL_MONITORING_ROOT);
185 if (threadPoolMonitoringObjectRoot == null) {
186 threadPoolMonitoringObjectRoot = MonitoringFactories.
187 getMonitoredObjectFactory().createMonitoredObject(
188 MonitoringConstants.THREADPOOL_MONITORING_ROOT,
189 MonitoringConstants.THREADPOOL_MONITORING_ROOT_DESCRIPTION);
190 root.addChild(threadPoolMonitoringObjectRoot);
191 }
192 threadpoolMonitoredObject = MonitoringFactories.
193 getMonitoredObjectFactory().
194 createMonitoredObject(name,
251
252 public WorkQueue getAnyWorkQueue()
253 {
254 return workQueue;
255 }
256
257 public WorkQueue getWorkQueue(int queueId)
258 throws NoSuchWorkQueueException
259 {
260 if (queueId != 0)
261 throw new NoSuchWorkQueueException();
262 return workQueue;
263 }
264
265 /**
266 * To be called from the workqueue when work is added to the
267 * workQueue. This method would create new threads if required
268 * or notify waiting threads on the queue for available work
269 */
270 void notifyForAvailableWork(WorkQueue aWorkQueue) {
271 synchronized (aWorkQueue) {
272 if (availableWorkerThreads < aWorkQueue.workItemsInQueue()) {
273 createWorkerThread();
274 } else {
275 aWorkQueue.notify();
276 }
277 }
278 }
279
280
281 private Thread createWorkerThreadHelper( String name ) {
282 // Thread creation needs to be in a doPrivileged block
283 // if there is a non-null security manager for two reasons:
284 // 1. The creation of a thread in a specific ThreadGroup
285 // is a privileged operation. Lack of a doPrivileged
286 // block here causes an AccessControlException
287 // (see bug 6268145).
288 // 2. We want to make sure that the permissions associated
289 // with this thread do NOT include the permissions of
290 // the current thread that is calling this method.
291 // This leads to problems in the app server where
292 // some threads in the ThreadPool randomly get
293 // bad permissions, leading to unpredictable
294 // permission errors (see bug 6021011).
295 //
296 // A Java thread contains a stack of call frames,
297 // one for each method called that has not yet returned.
298 // Each method comes from a particular class. The class
299 // was loaded by a ClassLoader which has an associated
300 // CodeSource, and this determines the Permissions
301 // for all methods in that class. The current
302 // Permissions for the thread are the intersection of
303 // all Permissions for the methods on the stack.
304 // This is part of the Security Context of the thread.
305 //
306 // When a thread creates a new thread, the new thread
307 // inherits the security context of the old thread.
308 // This is bad in a ThreadPool, because different
309 // creators of threads may have different security contexts.
310 // This leads to occasional unpredictable errors when
311 // a thread is re-used in a different security context.
312 //
313 // Avoiding this problem is simple: just do the thread
314 // creation in a doPrivileged block. This sets the
315 // inherited security context to that of the code source
316 // for the ORB code itself, which contains all permissions
317 // in either Java SE or Java EE.
318 WorkerThread thread = new WorkerThread(threadGroup, name);
319 synchronized (workersLock) {
320 workers.add(thread);
321 }
322
323 // The thread must be set to a daemon thread so the
324 // VM can exit if the only threads left are PooledThreads
325 // or other daemons. We don't want to rely on the
326 // calling thread always being a daemon.
327 // Note that no exception is possible here since we
328 // are inside the doPrivileged block.
329 thread.setDaemon(true);
330
331 wrapper.workerThreadCreated(thread, thread.getContextClassLoader());
332
333 thread.start();
334 return null;
335 }
336
337
338 /**
339 * To be called from the workqueue to create worker threads when none
340 * available.
341 */
342 void createWorkerThread() {
343 final String name = getName();
344 synchronized (workQueue) {
345 try {
346 if (System.getSecurityManager() == null) {
347 createWorkerThreadHelper(name);
348 } else {
349 // If we get here, we need to create a thread.
350 AccessController.doPrivileged(
351 new PrivilegedAction() {
352 public Object run() {
353 return createWorkerThreadHelper(name);
354 }
355 }
356 );
357 }
358 } catch (Throwable t) {
359 // Decrementing the count of current worker threads.
360 // But, it will be increased in the finally block.
361 decrementCurrentNumberOfThreads();
362 wrapper.workerThreadCreationFailure(t);
363 } finally {
364 incrementCurrentNumberOfThreads();
365 }
366 }
367 }
368
369 public int minimumNumberOfThreads() {
370 return minWorkerThreads;
371 }
372
373 public int maximumNumberOfThreads() {
374 return maxWorkerThreads;
375 }
376
377 public long idleTimeoutForThreads() {
378 return inactivityTimeout;
379 }
380
381 public int currentNumberOfThreads() {
382 synchronized (workQueue) {
383 return currentThreadCount;
384 }
385 }
386
387 void decrementCurrentNumberOfThreads() {
388 synchronized (workQueue) {
389 currentThreadCount--;
390 }
391 }
392
393 void incrementCurrentNumberOfThreads() {
394 synchronized (workQueue) {
395 currentThreadCount++;
396 }
397 }
398
399 public int numberOfAvailableThreads() {
400 synchronized (workQueue) {
401 return availableWorkerThreads;
402 }
403 }
404
405 public int numberOfBusyThreads() {
406 synchronized (workQueue) {
407 return (currentThreadCount - availableWorkerThreads);
408 }
409 }
410
411 public long averageWorkCompletionTime() {
412 synchronized (workQueue) {
413 return (totalTimeTaken.get() / processedCount.get());
414 }
415 }
416
417 public long currentProcessedCount() {
418 synchronized (workQueue) {
419 return processedCount.get();
420 }
421 }
422
423 public String getName() {
424 return name;
425 }
426
427 /**
428 * This method will return the number of WorkQueues serviced by the threadpool.
429 */
430 public int numberOfWorkQueues() {
431 return 1;
432 }
433
434
435 private static synchronized int getUniqueThreadId() {
436 return ThreadPoolImpl.threadCounter.incrementAndGet();
437 }
438
439 /**
440 * This method will decrement the number of available threads
441 * in the threadpool which are waiting for work. Called from
442 * WorkQueueImpl.requestWork()
443 */
444 void decrementNumberOfAvailableThreads() {
445 synchronized (workQueue) {
446 availableWorkerThreads--;
447 }
448 }
449
450 /**
451 * This method will increment the number of available threads
452 * in the threadpool which are waiting for work. Called from
453 * WorkQueueImpl.requestWork()
454 */
455 void incrementNumberOfAvailableThreads() {
456 synchronized (workQueue) {
457 availableWorkerThreads++;
458 }
459 }
460
461
462 private class WorkerThread extends Thread implements Closeable
463 {
464 private Work currentWork;
465 private int threadId = 0; // unique id for the thread
466 private volatile boolean closeCalled = false;
467 private String threadPoolName;
468 // name seen by Thread.getName()
469 private StringBuffer workerThreadName = new StringBuffer();
470
471 WorkerThread(ThreadGroup tg, String threadPoolName) {
472 super(tg, "Idle");
473 this.threadId = ThreadPoolImpl.getUniqueThreadId();
474 this.threadPoolName = threadPoolName;
475 setName(composeWorkerThreadName(threadPoolName, "Idle"));
476 }
477
478 public synchronized void close() {
479 closeCalled = true;
480 interrupt();
481 }
482
483 private void resetClassLoader() {
484
485 }
486
487 private void performWork() {
488 long start = System.currentTimeMillis();
489 try {
490 currentWork.doWork();
491 } catch (Throwable t) {
492 wrapper.workerThreadDoWorkThrowable(this, t);
493 }
494 long elapsedTime = System.currentTimeMillis() - start;
495 totalTimeTaken.addAndGet(elapsedTime);
496 processedCount.incrementAndGet();
497 }
498
499 public void run() {
500 try {
501 while (!closeCalled) {
502 try {
503 currentWork = ((WorkQueueImpl)workQueue).requestWork(
504 inactivityTimeout);
505 if (currentWork == null)
506 continue;
507 } catch (InterruptedException exc) {
508 wrapper.workQueueThreadInterrupted( exc, getName(),
509 Boolean.valueOf(closeCalled));
510
511 continue ;
512 } catch (Throwable t) {
513 wrapper.workerThreadThrowableFromRequestWork(this, t,
514 workQueue.getName());
515
516 continue;
517 }
518
519 performWork();
520
521 // set currentWork to null so that the work item can be
522 // garbage collected without waiting for the next work item.
523 currentWork = null;
524
525 resetClassLoader();
526 }
527 } catch (Throwable e) {
528 // This should not be possible
529 wrapper.workerThreadCaughtUnexpectedThrowable(this,e);
530 } finally {
531 synchronized (workersLock) {
532 workers.remove(this);
533 }
534 }
535 }
536
537 private String composeWorkerThreadName(String poolName, String workerName) {
538 workerThreadName.setLength(0);
539 workerThreadName.append("p: ").append(poolName);
540 workerThreadName.append("; w: ").append(workerName);
541 return workerThreadName.toString();
542 }
543 } // End of WorkerThread class
544
545 }
546
547 // End of file.
|