1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20 package org.apache.mina.core.service;
21
22 import java.lang.reflect.Constructor;
23 import java.nio.channels.spi.SelectorProvider;
24 import java.util.Arrays;
25 import java.util.concurrent.Executor;
26 import java.util.concurrent.ExecutorService;
27 import java.util.concurrent.Executors;
28 import java.util.concurrent.ThreadPoolExecutor;
29
30 import org.apache.mina.core.RuntimeIoException;
31 import org.apache.mina.core.session.AbstractIoSession;
32 import org.apache.mina.core.session.AttributeKey;
33 import org.apache.mina.core.session.IoSession;
34 import org.apache.mina.core.write.WriteRequest;
35 import org.slf4j.Logger;
36 import org.slf4j.LoggerFactory;
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80 public class SimpleIoProcessorPool<S extends AbstractIoSession> implements IoProcessor<S> {
81
82 private static final Logger LOGGER = LoggerFactory.getLogger(SimpleIoProcessorPool.class);
83
84
85 private static final int DEFAULT_SIZE = Runtime.getRuntime().availableProcessors() + 1;
86
87
88 private static final AttributeKeybuteKey.html#AttributeKey">AttributeKey PROCESSOR = new AttributeKey(SimpleIoProcessorPool.class, "processor");
89
90
91 private final IoProcessor<S>[] pool;
92
93
94 private final Executor executor;
95
96
97 private final boolean createdExecutor;
98
99
100 private final Object disposalLock = new Object();
101
102
103 private volatile boolean disposing;
104
105
106 private volatile boolean disposed;
107
108
109
110
111
112
113
114 public SimpleIoProcessorPool(Class<? extends IoProcessor<S>> processorType) {
115 this(processorType, null, DEFAULT_SIZE, null);
116 }
117
118
119
120
121
122
123
124
125 public SimpleIoProcessorPool(Class<? extends IoProcessor<S>> processorType, int size) {
126 this(processorType, null, size, null);
127 }
128
129
130
131
132
133
134
135
136
137 public SimpleIoProcessorPool(Class<? extends IoProcessor<S>> processorType, int size, SelectorProvider selectorProvider) {
138 this(processorType, null, size, selectorProvider);
139 }
140
141
142
143
144
145
146
147 public SimpleIoProcessorPool(Class<? extends IoProcessor<S>> processorType, Executor executor) {
148 this(processorType, executor, DEFAULT_SIZE, null);
149 }
150
151
152
153
154
155
156
157
158
159 @SuppressWarnings("unchecked")
160 public SimpleIoProcessorPool(Class<? extends IoProcessor<S>> processorType, Executor executor, int size,
161 SelectorProvider selectorProvider) {
162 if (processorType == null) {
163 throw new IllegalArgumentException("processorType");
164 }
165
166 if (size <= 0) {
167 throw new IllegalArgumentException("size: " + size + " (expected: positive integer)");
168 }
169
170
171 createdExecutor = executor == null;
172
173 if (createdExecutor) {
174 this.executor = Executors.newCachedThreadPool();
175
176 ((ThreadPoolExecutor) this.executor).setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy());
177 } else {
178 this.executor = executor;
179 }
180
181 pool = new IoProcessor[size];
182
183 boolean success = false;
184 Constructor<? extends IoProcessor<S>> processorConstructor = null;
185 boolean usesExecutorArg = true;
186
187 try {
188
189 try {
190 try {
191 processorConstructor = processorType.getConstructor(ExecutorService.class);
192 pool[0] = processorConstructor.newInstance(this.executor);
193 } catch (NoSuchMethodException e1) {
194
195 try {
196 if(selectorProvider==null) {
197 processorConstructor = processorType.getConstructor(Executor.class);
198 pool[0] = processorConstructor.newInstance(this.executor);
199 } else {
200 processorConstructor = processorType.getConstructor(Executor.class, SelectorProvider.class);
201 pool[0] = processorConstructor.newInstance(this.executor,selectorProvider);
202 }
203 } catch (NoSuchMethodException e2) {
204
205 try {
206 processorConstructor = processorType.getConstructor();
207 usesExecutorArg = false;
208 pool[0] = processorConstructor.newInstance();
209 } catch (NoSuchMethodException e3) {
210
211 }
212 }
213 }
214 } catch (RuntimeException re) {
215 LOGGER.error("Cannot create an IoProcessor :{}", re.getMessage());
216 throw re;
217 } catch (Exception e) {
218 String msg = "Failed to create a new instance of " + processorType.getName() + ":" + e.getMessage();
219 LOGGER.error(msg, e);
220 throw new RuntimeIoException(msg, e);
221 }
222
223 if (processorConstructor == null) {
224
225 String msg = String.valueOf(processorType) + " must have a public constructor with one "
226 + ExecutorService.class.getSimpleName() + " parameter, a public constructor with one "
227 + Executor.class.getSimpleName() + " parameter or a public default constructor.";
228 LOGGER.error(msg);
229 throw new IllegalArgumentException(msg);
230 }
231
232
233 for (int i = 1; i < pool.length; i++) {
234 try {
235 if (usesExecutorArg) {
236 if(selectorProvider==null) {
237 pool[i] = processorConstructor.newInstance(this.executor);
238 } else {
239 pool[i] = processorConstructor.newInstance(this.executor, selectorProvider);
240 }
241 } else {
242 pool[i] = processorConstructor.newInstance();
243 }
244 } catch (Exception e) {
245
246 }
247 }
248
249 success = true;
250 } finally {
251 if (!success) {
252 dispose();
253 }
254 }
255 }
256
257
258
259
260 @Override
261 public final void add(S session) {
262 getProcessor(session).add(session);
263 }
264
265
266
267
268 @Override
269 public final void flush(S session) {
270 getProcessor(session).flush(session);
271 }
272
273
274
275
276 @Override
277 public final void write(S session, WriteRequest writeRequest) {
278 getProcessor(session).write(session, writeRequest);
279 }
280
281
282
283
284 @Override
285 public final void remove(S session) {
286 getProcessor(session).remove(session);
287 }
288
289
290
291
292 @Override
293 public final void updateTrafficControl(S session) {
294 getProcessor(session).updateTrafficControl(session);
295 }
296
297
298
299
300 @Override
301 public boolean isDisposed() {
302 return disposed;
303 }
304
305
306
307
308 @Override
309 public boolean isDisposing() {
310 return disposing;
311 }
312
313
314
315
316 @Override
317 public final void dispose() {
318 if (disposed) {
319 return;
320 }
321
322 synchronized (disposalLock) {
323 if (!disposing) {
324 disposing = true;
325
326 for (IoProcessor<S> ioProcessor : pool) {
327 if (ioProcessor == null) {
328
329 continue;
330 }
331
332 if (ioProcessor.isDisposing()) {
333 continue;
334 }
335
336 try {
337 ioProcessor.dispose();
338 } catch (Exception e) {
339 LOGGER.warn("Failed to dispose the {} IoProcessor.", ioProcessor.getClass().getSimpleName(), e);
340 }
341 }
342
343 if (createdExecutor) {
344 ((ExecutorService) executor).shutdown();
345 }
346 }
347
348 Arrays.fill(pool, null);
349 disposed = true;
350 }
351 }
352
353
354
355
356
357 @SuppressWarnings("unchecked")
358 private IoProcessor<S> getProcessor(S session) {
359 IoProcessor<S> processor = (IoProcessor<S>) session.getAttribute(PROCESSOR);
360
361 if (processor == null) {
362 if (disposed || disposing) {
363 throw new IllegalStateException("A disposed processor cannot be accessed.");
364 }
365
366 processor = pool[Math.abs((int) session.getId()) % pool.length];
367
368 if (processor == null) {
369 throw new IllegalStateException("A disposed processor cannot be accessed.");
370 }
371
372 session.setAttributeIfAbsent(PROCESSOR, processor);
373 }
374
375 return processor;
376 }
377 }