|
Lines 78-83
Link Here
|
| 78 |
*/ |
78 |
*/ |
| 79 |
protected AtomicInteger submittedTasksCount; |
79 |
protected AtomicInteger submittedTasksCount; |
| 80 |
|
80 |
|
|
|
81 |
/** |
| 82 |
* The maximum number of elements that can queue up before we reject them |
| 83 |
*/ |
| 84 |
protected int maxQueueSize = Integer.MAX_VALUE; |
| 85 |
|
| 81 |
private LifecycleSupport lifecycle = new LifecycleSupport(this); |
86 |
private LifecycleSupport lifecycle = new LifecycleSupport(this); |
| 82 |
// ---------------------------------------------- Constructors |
87 |
// ---------------------------------------------- Constructors |
| 83 |
public StandardThreadExecutor() { |
88 |
public StandardThreadExecutor() { |
|
Lines 89-95
Link Here
|
| 89 |
// ---------------------------------------------- Public Methods |
94 |
// ---------------------------------------------- Public Methods |
| 90 |
public void start() throws LifecycleException { |
95 |
public void start() throws LifecycleException { |
| 91 |
lifecycle.fireLifecycleEvent(BEFORE_START_EVENT, null); |
96 |
lifecycle.fireLifecycleEvent(BEFORE_START_EVENT, null); |
| 92 |
TaskQueue taskqueue = new TaskQueue(); |
97 |
TaskQueue taskqueue = new TaskQueue(maxQueueSize); |
| 93 |
TaskThreadFactory tf = new TaskThreadFactory(namePrefix); |
98 |
TaskThreadFactory tf = new TaskThreadFactory(namePrefix); |
| 94 |
lifecycle.fireLifecycleEvent(START_EVENT, null); |
99 |
lifecycle.fireLifecycleEvent(START_EVENT, null); |
| 95 |
executor = new ThreadPoolExecutor(getMinSpareThreads(), getMaxThreads(), maxIdleTime, TimeUnit.MILLISECONDS,taskqueue, tf) { |
100 |
executor = new ThreadPoolExecutor(getMinSpareThreads(), getMaxThreads(), maxIdleTime, TimeUnit.MILLISECONDS,taskqueue, tf) { |
|
Lines 114-120
Link Here
|
| 114 |
submittedTasksCount = null; |
119 |
submittedTasksCount = null; |
| 115 |
lifecycle.fireLifecycleEvent(AFTER_STOP_EVENT, null); |
120 |
lifecycle.fireLifecycleEvent(AFTER_STOP_EVENT, null); |
| 116 |
} |
121 |
} |
| 117 |
|
122 |
|
|
|
123 |
// This method is not used by Tomcat 6, but is available in later versions |
| 124 |
public void execute(Runnable command, long timeout, TimeUnit unit) { |
| 125 |
if ( executor != null ) { |
| 126 |
submittedTasksCount.incrementAndGet(); |
| 127 |
try { |
| 128 |
executor.execute(command); |
| 129 |
} catch (RejectedExecutionException rx) { |
| 130 |
//there could have been contention around the queue |
| 131 |
try { |
| 132 |
if ( !( (TaskQueue) executor.getQueue()).force(command,timeout,unit) ) { |
| 133 |
submittedTasksCount.decrementAndGet(); |
| 134 |
throw new RejectedExecutionException("Work queue full."); |
| 135 |
} |
| 136 |
}catch (InterruptedException x) { |
| 137 |
throw new RejectedExecutionException("Interrupted.",x); |
| 138 |
} |
| 139 |
} |
| 140 |
} else throw new IllegalStateException("StandardThreadPool not started."); |
| 141 |
} |
| 142 |
|
| 118 |
public void execute(Runnable command) { |
143 |
public void execute(Runnable command) { |
| 119 |
if ( executor != null ) { |
144 |
if ( executor != null ) { |
| 120 |
submittedTasksCount.incrementAndGet(); |
145 |
submittedTasksCount.incrementAndGet(); |
|
Lines 124-130
Link Here
|
| 124 |
//there could have been contention around the queue |
149 |
//there could have been contention around the queue |
| 125 |
if ( !( (TaskQueue) executor.getQueue()).force(command) ) { |
150 |
if ( !( (TaskQueue) executor.getQueue()).force(command) ) { |
| 126 |
submittedTasksCount.decrementAndGet(); |
151 |
submittedTasksCount.decrementAndGet(); |
| 127 |
throw new RejectedExecutionException(); |
152 |
throw new RejectedExecutionException("Work queue full."); |
| 128 |
} |
153 |
} |
| 129 |
} |
154 |
} |
| 130 |
} else throw new IllegalStateException("StandardThreadPool not started."); |
155 |
} else throw new IllegalStateException("StandardThreadPool not started."); |
|
Lines 196-201
Link Here
|
| 196 |
this.name = name; |
221 |
this.name = name; |
| 197 |
} |
222 |
} |
| 198 |
|
223 |
|
|
|
224 |
public void setMaxQueueSize(int size) { |
| 225 |
this.maxQueueSize = size; |
| 226 |
} |
| 227 |
|
| 228 |
public int getMaxQueueSize() { |
| 229 |
return maxQueueSize; |
| 230 |
} |
| 231 |
|
| 199 |
/** |
232 |
/** |
| 200 |
* Add a LifecycleEvent listener to this component. |
233 |
* Add a LifecycleEvent listener to this component. |
| 201 |
* |
234 |
* |
|
Lines 257-264
Link Here
|
| 257 |
super(); |
290 |
super(); |
| 258 |
} |
291 |
} |
| 259 |
|
292 |
|
| 260 |
public TaskQueue(int initialCapacity) { |
293 |
public TaskQueue(int capacity) { |
| 261 |
super(initialCapacity); |
294 |
super(capacity); |
| 262 |
} |
295 |
} |
| 263 |
|
296 |
|
| 264 |
public TaskQueue(Collection<? extends Runnable> c) { |
297 |
public TaskQueue(Collection<? extends Runnable> c) { |
|
Lines 274-279
Link Here
|
| 274 |
return super.offer(o); //forces the item onto the queue, to be used if the task is rejected |
307 |
return super.offer(o); //forces the item onto the queue, to be used if the task is rejected |
| 275 |
} |
308 |
} |
| 276 |
|
309 |
|
|
|
310 |
public boolean force(Runnable o, long timeout, TimeUnit unit) throws InterruptedException { |
| 311 |
if ( parent.isShutdown() ) throw new RejectedExecutionException("Executor not running, can't force a command into the queue"); |
| 312 |
return super.offer(o,timeout,unit); //forces the item onto the queue, to be used if the task is rejected |
| 313 |
} |
| 314 |
|
| 277 |
public boolean offer(Runnable o) { |
315 |
public boolean offer(Runnable o) { |
| 278 |
//we can't do any checks |
316 |
//we can't do any checks |
| 279 |
if (parent==null) return super.offer(o); |
317 |
if (parent==null) return super.offer(o); |