1 /*
2 * Licensed to the Apache Software Foundation (ASF) under one
3 * or more contributor license agreements. See the NOTICE file
4 * distributed with this work for additional information
5 * regarding copyright ownership. The ASF licenses this file
6 * to you under the Apache License, Version 2.0 (the
7 * "License"); you may not use this file except in compliance
8 * with the License. You may obtain a copy of the License at
9 *
10 * http://www.apache.org/licenses/LICENSE-2.0
11 *
12 * Unless required by applicable law or agreed to in writing,
13 * software distributed under the License is distributed on an
14 * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
15 * KIND, either express or implied. See the License for the
16 * specific language governing permissions and limitations
17 * under the License.
18 *
19 */
20 package org.apache.mina.transport.socket.nio;
21
22 import java.io.IOException;
23 import java.nio.channels.SelectionKey;
24 import java.nio.channels.Selector;
25 import java.nio.channels.SocketChannel;
26 import java.util.Queue;
27 import java.util.Set;
28 import java.util.concurrent.ConcurrentLinkedQueue;
29 import java.util.concurrent.Executor;
30
31 import org.apache.mina.common.ByteBuffer;
32 import org.apache.mina.common.ExceptionMonitor;
33 import org.apache.mina.common.IdleStatus;
34 import org.apache.mina.common.IoSession;
35 import org.apache.mina.common.WriteTimeoutException;
36 import org.apache.mina.common.IoFilter.WriteRequest;
37 import org.apache.mina.util.NamePreservingRunnable;
38
39 /**
40 * Performs all I/O operations for sockets which is connected or bound. This class is used by MINA internally.
41 *
42 * @author The Apache Directory Project (mina-dev@directory.apache.org)
43 * @version $Rev: 592224 $, $Date: 2007-11-06 11:11:40 +0900 (Tue, 06 Nov 2007) $,
44 */
45 class SocketIoProcessor {
46
47 /**
48 * The maximum loop count for a write operation until
49 * {@link #write(IoSession, IoBuffer)} returns non-zero value.
50 * It is similar to what a spin lock is for in concurrency programming.
51 * It improves memory utilization and write throughput significantly.
52 */
53 private static final int WRITE_SPIN_COUNT = 256;
54
55 private final Object lock = new Object();
56
57 private final String threadName;
58
59 private final Executor executor;
60
61 private volatile Selector selector;
62
63 private final Queue<SocketSessionImpl> newSessions = new ConcurrentLinkedQueue<SocketSessionImpl>();
64
65 private final Queue<SocketSessionImpl> removingSessions = new ConcurrentLinkedQueue<SocketSessionImpl>();
66
67 private final Queue<SocketSessionImpl> flushingSessions = new ConcurrentLinkedQueue<SocketSessionImpl>();
68
69 private final Queue<SocketSessionImpl> trafficControllingSessions = new ConcurrentLinkedQueue<SocketSessionImpl>();
70
71 private Worker worker;
72
73 private long lastIdleCheckTime = System.currentTimeMillis();
74
75 SocketIoProcessor(String threadName, Executor executor) {
76 this.threadName = threadName;
77 this.executor = executor;
78 }
79
80 void addNew(SocketSessionImpl session) throws IOException {
81 newSessions.add(session);
82 startupWorker();
83 }
84
85 void remove(SocketSessionImpl session) throws IOException {
86 scheduleRemove(session);
87 startupWorker();
88 }
89
90 private void startupWorker() throws IOException {
91 synchronized (lock) {
92 if (worker == null) {
93 selector = Selector.open();
94 worker = new Worker();
95 executor.execute(new NamePreservingRunnable(worker, threadName));
96 }
97 selector.wakeup();
98 }
99 }
100
101 void flush(SocketSessionImpl session) {
102 if ( scheduleFlush(session) ) {
103 Selector selector = this.selector;
104 if (selector != null) {
105 selector.wakeup();
106 }
107 }
108 }
109
110 void updateTrafficMask(SocketSessionImpl session) {
111 scheduleTrafficControl(session);
112 Selector selector = this.selector;
113 if (selector != null) {
114 selector.wakeup();
115 }
116 }
117
118 private void scheduleRemove(SocketSessionImpl session) {
119 removingSessions.add(session);
120 }
121
122 private boolean scheduleFlush(SocketSessionImpl session) {
123 if (session.setScheduledForFlush(true)) {
124 flushingSessions.add(session);
125
126 return true;
127 }
128
129 return false;
130 }
131
132 private void scheduleTrafficControl(SocketSessionImpl session) {
133 trafficControllingSessions.add(session);
134 }
135
136 private void doAddNew() {
137 Selector selector = this.selector;
138 for (;;) {
139 SocketSessionImpl session = newSessions.poll();
140
141 if (session == null)
142 break;
143
144 SocketChannel ch = session.getChannel();
145 try {
146 ch.configureBlocking(false);
147 session.setSelectionKey(ch.register(selector,
148 SelectionKey.OP_READ, session));
149
150 // AbstractIoFilterChain.CONNECT_FUTURE is cleared inside here
151 // in AbstractIoFilterChain.fireSessionOpened().
152 session.getServiceListeners().fireSessionCreated(session);
153 } catch (IOException e) {
154 // Clear the AbstractIoFilterChain.CONNECT_FUTURE attribute
155 // and call ConnectFuture.setException().
156 session.getFilterChain().fireExceptionCaught(session, e);
157 }
158 }
159 }
160
161 private void doRemove() {
162 for (;;) {
163 SocketSessionImpl session = removingSessions.poll();
164
165 if (session == null)
166 break;
167
168 SocketChannel ch = session.getChannel();
169 SelectionKey key = session.getSelectionKey();
170 // Retry later if session is not yet fully initialized.
171 // (In case that Session.close() is called before addSession() is processed)
172 if (key == null) {
173 scheduleRemove(session);
174 break;
175 }
176 // skip if channel is already closed
177 if (!key.isValid()) {
178 continue;
179 }
180
181 try {
182 key.cancel();
183 ch.close();
184 } catch (IOException e) {
185 session.getFilterChain().fireExceptionCaught(session, e);
186 } finally {
187 releaseWriteBuffers(session);
188 session.getServiceListeners().fireSessionDestroyed(session);
189 }
190 }
191 }
192
193 private void process(Set<SelectionKey> selectedKeys) {
194 for (SelectionKey key : selectedKeys) {
195 SocketSessionImpl session = (SocketSessionImpl) key.attachment();
196
197 if (key.isReadable() && session.getTrafficMask().isReadable()) {
198 read(session);
199 }
200
201 if (key.isWritable() && session.getTrafficMask().isWritable()) {
202 scheduleFlush(session);
203 }
204 }
205
206 selectedKeys.clear();
207 }
208
209 private void read(SocketSessionImpl session) {
210 ByteBuffer buf = ByteBuffer.allocate(session.getReadBufferSize());
211 SocketChannel ch = session.getChannel();
212
213 try {
214 int readBytes = 0;
215 int ret;
216
217 try {
218 while ((ret = ch.read(buf.buf())) > 0) {
219 readBytes += ret;
220 }
221 } finally {
222 buf.flip();
223 }
224
225 session.increaseReadBytes(readBytes);
226
227 if (readBytes > 0) {
228 session.getFilterChain().fireMessageReceived(session, buf);
229 buf = null;
230
231 if (readBytes * 2 < session.getReadBufferSize()) {
232 session.decreaseReadBufferSize();
233 } else if (readBytes == session.getReadBufferSize()) {
234 session.increaseReadBufferSize();
235 }
236 }
237 if (ret < 0) {
238 scheduleRemove(session);
239 }
240 } catch (Throwable e) {
241 if (e instanceof IOException)
242 scheduleRemove(session);
243 session.getFilterChain().fireExceptionCaught(session, e);
244 } finally {
245 if (buf != null)
246 buf.release();
247 }
248 }
249
250 private void notifyIdleness() {
251 // process idle sessions
252 long currentTime = System.currentTimeMillis();
253 if ((currentTime - lastIdleCheckTime) >= 1000) {
254 lastIdleCheckTime = currentTime;
255 Set<SelectionKey> keys = selector.keys();
256 if (keys != null) {
257 for (SelectionKey key : keys) {
258 SocketSessionImpl session = (SocketSessionImpl) key
259 .attachment();
260 notifyIdleness(session, currentTime);
261 }
262 }
263 }
264 }
265
266 private void notifyIdleness(SocketSessionImpl session, long currentTime) {
267 notifyIdleness0(session, currentTime, session
268 .getIdleTimeInMillis(IdleStatus.BOTH_IDLE),
269 IdleStatus.BOTH_IDLE, Math.max(session.getLastIoTime(), session
270 .getLastIdleTime(IdleStatus.BOTH_IDLE)));
271 notifyIdleness0(session, currentTime, session
272 .getIdleTimeInMillis(IdleStatus.READER_IDLE),
273 IdleStatus.READER_IDLE, Math.max(session.getLastReadTime(),
274 session.getLastIdleTime(IdleStatus.READER_IDLE)));
275 notifyIdleness0(session, currentTime, session
276 .getIdleTimeInMillis(IdleStatus.WRITER_IDLE),
277 IdleStatus.WRITER_IDLE, Math.max(session.getLastWriteTime(),
278 session.getLastIdleTime(IdleStatus.WRITER_IDLE)));
279
280 notifyWriteTimeout(session, currentTime, session
281 .getWriteTimeoutInMillis(), session.getLastWriteTime());
282 }
283
284 private void notifyIdleness0(SocketSessionImpl session, long currentTime,
285 long idleTime, IdleStatus status, long lastIoTime) {
286 if (idleTime > 0 && lastIoTime != 0
287 && (currentTime - lastIoTime) >= idleTime) {
288 session.increaseIdleCount(status);
289 session.getFilterChain().fireSessionIdle(session, status);
290 }
291 }
292
293 private void notifyWriteTimeout(SocketSessionImpl session,
294 long currentTime, long writeTimeout, long lastIoTime) {
295 SelectionKey key = session.getSelectionKey();
296 if (writeTimeout > 0 && (currentTime - lastIoTime) >= writeTimeout
297 && key != null && key.isValid()
298 && (key.interestOps() & SelectionKey.OP_WRITE) != 0) {
299 session.getFilterChain().fireExceptionCaught(session,
300 new WriteTimeoutException());
301 }
302 }
303
304 private void doFlush() {
305 for (;;) {
306 SocketSessionImpl session = flushingSessions.poll();
307
308 if (session == null)
309 break;
310
311 session.setScheduledForFlush(false);
312
313 if (!session.isConnected()) {
314 releaseWriteBuffers(session);
315 continue;
316 }
317
318 SelectionKey key = session.getSelectionKey();
319 // Retry later if session is not yet fully initialized.
320 // (In case that Session.write() is called before addSession() is processed)
321 if (key == null) {
322 scheduleFlush(session);
323 break;
324 }
325
326 // Skip if the channel is already closed.
327 if (!key.isValid()) {
328 continue;
329 }
330
331 try {
332 boolean flushedAll = doFlush(session);
333 if( flushedAll && !session.getWriteRequestQueue().isEmpty() && !session.isScheduledForFlush()) {
334 scheduleFlush( session );
335 }
336 } catch (IOException e) {
337 scheduleRemove(session);
338 session.getFilterChain().fireExceptionCaught(session, e);
339 }
340 }
341 }
342
343 private void releaseWriteBuffers(SocketSessionImpl session) {
344 Queue<WriteRequest> writeRequestQueue = session.getWriteRequestQueue();
345 WriteRequest req;
346
347 if ((req = writeRequestQueue.poll()) != null) {
348 ByteBuffer buf = (ByteBuffer) req.getMessage();
349 try {
350 buf.release();
351 } catch (IllegalStateException e) {
352 session.getFilterChain().fireExceptionCaught(session, e);
353 } finally {
354 // The first unwritten empty buffer must be
355 // forwarded to the filter chain.
356 if (buf.hasRemaining()) {
357 req.getFuture().setWritten(false);
358 } else {
359 session.getFilterChain().fireMessageSent(session, req);
360 }
361 }
362
363 // Discard others.
364 while ((req = writeRequestQueue.poll()) != null) {
365 try {
366 ((ByteBuffer) req.getMessage()).release();
367 } catch (IllegalStateException e) {
368 session.getFilterChain().fireExceptionCaught(session, e);
369 } finally {
370 req.getFuture().setWritten(false);
371 }
372 }
373 }
374 }
375
376 private boolean doFlush(SocketSessionImpl session) throws IOException {
377 SocketChannel ch = session.getChannel();
378 if (!ch.isConnected()) {
379 scheduleRemove(session);
380 return false;
381 }
382
383 // Clear OP_WRITE
384 SelectionKey key = session.getSelectionKey();
385 key.interestOps(key.interestOps() & (~SelectionKey.OP_WRITE));
386
387 Queue<WriteRequest> writeRequestQueue = session.getWriteRequestQueue();
388
389 int writtenBytes = 0;
390 int maxWrittenBytes = ((SocketSessionConfig) session.getConfig()).getSendBufferSize() << 1;
391 try {
392 for (;;) {
393 WriteRequest req = writeRequestQueue.peek();
394
395 if (req == null)
396 break;
397
398 ByteBuffer buf = (ByteBuffer) req.getMessage();
399 if (buf.remaining() == 0) {
400 writeRequestQueue.poll();
401
402 buf.reset();
403
404 if (!buf.hasRemaining()) {
405 session.increaseWrittenMessages();
406 }
407
408 session.getFilterChain().fireMessageSent(session, req);
409 continue;
410 }
411
412 int localWrittenBytes = 0;
413 for (int i = WRITE_SPIN_COUNT; i > 0; i --) {
414 localWrittenBytes = ch.write(buf.buf());
415 if (localWrittenBytes != 0 || !buf.hasRemaining()) {
416 break;
417 }
418 }
419
420 writtenBytes += localWrittenBytes;
421
422 if (localWrittenBytes == 0 || writtenBytes >= maxWrittenBytes) {
423 // Kernel buffer is full or wrote too much.
424 key.interestOps(key.interestOps() | SelectionKey.OP_WRITE);
425 return false;
426 }
427 }
428 } finally {
429 session.increaseWrittenBytes(writtenBytes);
430 }
431
432 return true;
433 }
434
435 private void doUpdateTrafficMask() {
436 if (trafficControllingSessions.isEmpty())
437 return;
438
439 for (;;) {
440 SocketSessionImpl session = trafficControllingSessions.poll();
441
442 if (session == null)
443 break;
444
445 SelectionKey key = session.getSelectionKey();
446 // Retry later if session is not yet fully initialized.
447 // (In case that Session.suspend??() or session.resume??() is
448 // called before addSession() is processed)
449 if (key == null) {
450 scheduleTrafficControl(session);
451 break;
452 }
453 // skip if channel is already closed
454 if (!key.isValid()) {
455 continue;
456 }
457
458 // The normal is OP_READ and, if there are write requests in the
459 // session's write queue, set OP_WRITE to trigger flushing.
460 int ops = SelectionKey.OP_READ;
461 Queue<WriteRequest> writeRequestQueue = session
462 .getWriteRequestQueue();
463 synchronized (writeRequestQueue) {
464 if (!writeRequestQueue.isEmpty()) {
465 ops |= SelectionKey.OP_WRITE;
466 }
467 }
468
469 // Now mask the preferred ops with the mask of the current session
470 int mask = session.getTrafficMask().getInterestOps();
471 key.interestOps(ops & mask);
472 }
473 }
474
475 private class Worker implements Runnable {
476 public void run() {
477 Selector selector = SocketIoProcessor.this.selector;
478 for (;;) {
479 try {
480 int nKeys = selector.select(1000);
481 doAddNew();
482 doUpdateTrafficMask();
483
484 if (nKeys > 0) {
485 process(selector.selectedKeys());
486 }
487
488 doFlush();
489 doRemove();
490 notifyIdleness();
491
492 if (selector.keys().isEmpty()) {
493 synchronized (lock) {
494 if (selector.keys().isEmpty()
495 && newSessions.isEmpty()) {
496 worker = null;
497
498 try {
499 selector.close();
500 } catch (IOException e) {
501 ExceptionMonitor.getInstance()
502 .exceptionCaught(e);
503 } finally {
504 selector = null;
505 }
506
507 break;
508 }
509 }
510 }
511 } catch (Throwable t) {
512 ExceptionMonitor.getInstance().exceptionCaught(t);
513
514 try {
515 Thread.sleep(1000);
516 } catch (InterruptedException e1) {
517 ExceptionMonitor.getInstance().exceptionCaught(e1);
518 }
519 }
520 }
521 }
522 }
523 }
以上源码来自： 即时通讯网 - 即时通讯开发者社区！