View Javadoc

1   /*
2    * Copyright 2009 Red Hat, Inc.
3    *
4    * Red Hat licenses this file to you under the Apache License, version 2.0
5    * (the "License"); you may not use this file except in compliance with the
6    * License.  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, WITHOUT
12   * WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.  See the
13   * License for the specific language governing permissions and limitations
14   * under the License.
15   */
16  package org.jboss.netty.channel.socket.nio;
17  
18  import static org.jboss.netty.channel.Channels.*;
19  
20  import java.io.IOException;
21  import java.net.SocketAddress;
22  import java.nio.ByteBuffer;
23  import java.nio.channels.AsynchronousCloseException;
24  import java.nio.channels.CancelledKeyException;
25  import java.nio.channels.ClosedChannelException;
26  import java.nio.channels.NotYetConnectedException;
27  import java.nio.channels.SelectionKey;
28  import java.nio.channels.Selector;
29  import java.nio.channels.SocketChannel;
30  import java.util.Iterator;
31  import java.util.Queue;
32  import java.util.Set;
33  import java.util.concurrent.Executor;
34  import java.util.concurrent.ExecutorService;
35  import java.util.concurrent.atomic.AtomicBoolean;
36  import java.util.concurrent.locks.ReadWriteLock;
37  import java.util.concurrent.locks.ReentrantReadWriteLock;
38  
39  import org.jboss.netty.buffer.ChannelBuffer;
40  import org.jboss.netty.buffer.ChannelBufferFactory;
41  import org.jboss.netty.channel.Channel;
42  import org.jboss.netty.channel.ChannelException;
43  import org.jboss.netty.channel.ChannelFuture;
44  import org.jboss.netty.channel.MessageEvent;
45  import org.jboss.netty.channel.ReceiveBufferSizePredictor;
46  import org.jboss.netty.channel.socket.nio.SocketSendBufferPool.SendBuffer;
47  import org.jboss.netty.logging.InternalLogger;
48  import org.jboss.netty.logging.InternalLoggerFactory;
49  import org.jboss.netty.util.ThreadRenamingRunnable;
50  import org.jboss.netty.util.internal.IoWorkerRunnable;
51  import org.jboss.netty.util.internal.LinkedTransferQueue;
52  
53  /**
54   *
55   * @author <a href="http://www.jboss.org/netty/">The Netty Project</a>
56   * @author <a href="http://gleamynode.net/">Trustin Lee</a>
57   *
58   * @version $Rev: 2376 $, $Date: 2010-10-25 03:24:20 +0900 (Mon, 25 Oct 2010) $
59   *
60   */
61  class NioWorker implements Runnable {
62  
63      private static final InternalLogger logger =
64          InternalLoggerFactory.getInstance(NioWorker.class);
65  
66      private static final int CONSTRAINT_LEVEL = NioProviderMetadata.CONSTRAINT_LEVEL;
67  
68      static final int CLEANUP_INTERVAL = 256; // XXX Hard-coded value, but won't need customization.
69  
70      private final int bossId;
71      private final int id;
72      private final Executor executor;
73      private boolean started;
74      private volatile Thread thread;
75      volatile Selector selector;
76      private final AtomicBoolean wakenUp = new AtomicBoolean();
77      private final ReadWriteLock selectorGuard = new ReentrantReadWriteLock();
78      private final Object startStopLock = new Object();
79      private final Queue<Runnable> registerTaskQueue = new LinkedTransferQueue<Runnable>();
80      private final Queue<Runnable> writeTaskQueue = new LinkedTransferQueue<Runnable>();
81      private volatile int cancelledKeys; // should use AtomicInteger but we just need approximation
82  
83      private final SocketReceiveBufferPool recvBufferPool = new SocketReceiveBufferPool();
84      private final SocketSendBufferPool sendBufferPool = new SocketSendBufferPool();
85  
86      NioWorker(int bossId, int id, Executor executor) {
87          this.bossId = bossId;
88          this.id = id;
89          this.executor = executor;
90      }
91  
92      void register(NioSocketChannel channel, ChannelFuture future) {
93  
94          boolean server = !(channel instanceof NioClientSocketChannel);
95          Runnable registerTask = new RegisterTask(channel, future, server);
96          Selector selector;
97  
98          synchronized (startStopLock) {
99              if (!started) {
100                 // Open a selector if this worker didn't start yet.
101                 try {
102                     this.selector = selector = Selector.open();
103                 } catch (Throwable t) {
104                     throw new ChannelException(
105                             "Failed to create a selector.", t);
106                 }
107 
108                 // Start the worker thread with the new Selector.
109                 String threadName =
110                     (server ? "New I/O server worker #"
111                             : "New I/O client worker #") + bossId + '-' + id;
112 
113                 boolean success = false;
114                 try {
115                     executor.execute(
116                             new IoWorkerRunnable(
117                                     new ThreadRenamingRunnable(this, threadName)));
118                     success = true;
119                 } finally {
120                     if (!success) {
121                         // Release the Selector if the execution fails.
122                         try {
123                             selector.close();
124                         } catch (Throwable t) {
125                             logger.warn("Failed to close a selector.", t);
126                         }
127                         this.selector = selector = null;
128                         // The method will return to the caller at this point.
129                     }
130                 }
131             } else {
132                 // Use the existing selector if this worker has been started.
133                 selector = this.selector;
134             }
135 
136             assert selector != null && selector.isOpen();
137 
138             started = true;
139             boolean offered = registerTaskQueue.offer(registerTask);
140             assert offered;
141         }
142 
143         if (wakenUp.compareAndSet(false, true)) {
144             selector.wakeup();
145         }
146     }
147 
148     public void run() {
149         thread = Thread.currentThread();
150 
151         boolean shutdown = false;
152         Selector selector = this.selector;
153         for (;;) {
154             wakenUp.set(false);
155 
156             if (CONSTRAINT_LEVEL != 0) {
157                 selectorGuard.writeLock().lock();
158                     // This empty synchronization block prevents the selector
159                     // from acquiring its lock.
160                 selectorGuard.writeLock().unlock();
161             }
162 
163             try {
164                 SelectorUtil.select(selector);
165 
166                 // 'wakenUp.compareAndSet(false, true)' is always evaluated
167                 // before calling 'selector.wakeup()' to reduce the wake-up
168                 // overhead. (Selector.wakeup() is an expensive operation.)
169                 //
170                 // However, there is a race condition in this approach.
171                 // The race condition is triggered when 'wakenUp' is set to
172                 // true too early.
173                 //
174                 // 'wakenUp' is set to true too early if:
175                 // 1) Selector is waken up between 'wakenUp.set(false)' and
176                 //    'selector.select(...)'. (BAD)
177                 // 2) Selector is waken up between 'selector.select(...)' and
178                 //    'if (wakenUp.get()) { ... }'. (OK)
179                 //
180                 // In the first case, 'wakenUp' is set to true and the
181                 // following 'selector.select(...)' will wake up immediately.
182                 // Until 'wakenUp' is set to false again in the next round,
183                 // 'wakenUp.compareAndSet(false, true)' will fail, and therefore
184                 // any attempt to wake up the Selector will fail, too, causing
185                 // the following 'selector.select(...)' call to block
186                 // unnecessarily.
187                 //
188                 // To fix this problem, we wake up the selector again if wakenUp
189                 // is true immediately after selector.select(...).
190                 // It is inefficient in that it wakes up the selector for both
191                 // the first case (BAD - wake-up required) and the second case
192                 // (OK - no wake-up required).
193 
194                 if (wakenUp.get()) {
195                     selector.wakeup();
196                 }
197 
198                 cancelledKeys = 0;
199                 processRegisterTaskQueue();
200                 processWriteTaskQueue();
201                 processSelectedKeys(selector.selectedKeys());
202 
203                 // Exit the loop when there's nothing to handle.
204                 // The shutdown flag is used to delay the shutdown of this
205                 // loop to avoid excessive Selector creation when
206                 // connections are registered in a one-by-one manner instead of
207                 // concurrent manner.
208                 if (selector.keys().isEmpty()) {
209                     if (shutdown ||
210                         executor instanceof ExecutorService && ((ExecutorService) executor).isShutdown()) {
211 
212                         synchronized (startStopLock) {
213                             if (registerTaskQueue.isEmpty() && selector.keys().isEmpty()) {
214                                 started = false;
215                                 try {
216                                     selector.close();
217                                 } catch (IOException e) {
218                                     logger.warn(
219                                             "Failed to close a selector.", e);
220                                 } finally {
221                                     this.selector = null;
222                                 }
223                                 break;
224                             } else {
225                                 shutdown = false;
226                             }
227                         }
228                     } else {
229                         // Give one more second.
230                         shutdown = true;
231                     }
232                 } else {
233                     shutdown = false;
234                 }
235             } catch (Throwable t) {
236                 logger.warn(
237                         "Unexpected exception in the selector loop.", t);
238 
239                 // Prevent possible consecutive immediate failures that lead to
240                 // excessive CPU consumption.
241                 try {
242                     Thread.sleep(1000);
243                 } catch (InterruptedException e) {
244                     // Ignore.
245                 }
246             }
247         }
248     }
249 
250     private void processRegisterTaskQueue() throws IOException {
251         for (;;) {
252             final Runnable task = registerTaskQueue.poll();
253             if (task == null) {
254                 break;
255             }
256 
257             task.run();
258             cleanUpCancelledKeys();
259         }
260     }
261 
262     private void processWriteTaskQueue() throws IOException {
263         for (;;) {
264             final Runnable task = writeTaskQueue.poll();
265             if (task == null) {
266                 break;
267             }
268 
269             task.run();
270             cleanUpCancelledKeys();
271         }
272     }
273 
274     private void processSelectedKeys(Set<SelectionKey> selectedKeys) throws IOException {
275         for (Iterator<SelectionKey> i = selectedKeys.iterator(); i.hasNext();) {
276             SelectionKey k = i.next();
277             i.remove();
278             try {
279                 int readyOps = k.readyOps();
280                 if ((readyOps & SelectionKey.OP_READ) != 0 || readyOps == 0) {
281                     if (!read(k)) {
282                         // Connection already closed - no need to handle write.
283                         continue;
284                     }
285                 }
286                 if ((readyOps & SelectionKey.OP_WRITE) != 0) {
287                     writeFromSelectorLoop(k);
288                 }
289             } catch (CancelledKeyException e) {
290                 close(k);
291             }
292 
293             if (cleanUpCancelledKeys()) {
294                 break; // break the loop to avoid ConcurrentModificationException
295             }
296         }
297     }
298 
299     private boolean cleanUpCancelledKeys() throws IOException {
300         if (cancelledKeys >= CLEANUP_INTERVAL) {
301             cancelledKeys = 0;
302             selector.selectNow();
303             return true;
304         }
305         return false;
306     }
307 
308     private boolean read(SelectionKey k) {
309         final SocketChannel ch = (SocketChannel) k.channel();
310         final NioSocketChannel channel = (NioSocketChannel) k.attachment();
311 
312         final ReceiveBufferSizePredictor predictor =
313             channel.getConfig().getReceiveBufferSizePredictor();
314         final int predictedRecvBufSize = predictor.nextReceiveBufferSize();
315 
316         int ret = 0;
317         int readBytes = 0;
318         boolean failure = true;
319 
320         ByteBuffer bb = recvBufferPool.acquire(predictedRecvBufSize);
321         try {
322             while ((ret = ch.read(bb)) > 0) {
323                 readBytes += ret;
324                 if (!bb.hasRemaining()) {
325                     break;
326                 }
327             }
328             failure = false;
329         } catch (ClosedChannelException e) {
330             // Can happen, and does not need a user attention.
331         } catch (Throwable t) {
332             fireExceptionCaught(channel, t);
333         }
334 
335         if (readBytes > 0) {
336             bb.flip();
337 
338             final ChannelBufferFactory bufferFactory =
339                 channel.getConfig().getBufferFactory();
340             final ChannelBuffer buffer = bufferFactory.getBuffer(readBytes);
341             buffer.setBytes(0, bb);
342             buffer.writerIndex(readBytes);
343 
344             recvBufferPool.release(bb);
345 
346             // Update the predictor.
347             predictor.previousReceiveBufferSize(readBytes);
348 
349             // Fire the event.
350             fireMessageReceived(channel, buffer);
351         } else {
352             recvBufferPool.release(bb);
353         }
354 
355         if (ret < 0 || failure) {
356             close(channel, succeededFuture(channel));
357             return false;
358         }
359 
360         return true;
361     }
362 
363     private void close(SelectionKey k) {
364         NioSocketChannel ch = (NioSocketChannel) k.attachment();
365         close(ch, succeededFuture(ch));
366     }
367 
368     void writeFromUserCode(final NioSocketChannel channel) {
369         if (!channel.isConnected()) {
370             cleanUpWriteBuffer(channel);
371             return;
372         }
373 
374         if (scheduleWriteIfNecessary(channel)) {
375             return;
376         }
377 
378         // From here, we are sure Thread.currentThread() == workerThread.
379 
380         if (channel.writeSuspended) {
381             return;
382         }
383 
384         if (channel.inWriteNowLoop) {
385             return;
386         }
387 
388         write0(channel);
389     }
390 
391     void writeFromTaskLoop(final NioSocketChannel ch) {
392         if (!ch.writeSuspended) {
393             write0(ch);
394         }
395     }
396 
397     void writeFromSelectorLoop(final SelectionKey k) {
398         NioSocketChannel ch = (NioSocketChannel) k.attachment();
399         ch.writeSuspended = false;
400         write0(ch);
401     }
402 
403     private boolean scheduleWriteIfNecessary(final NioSocketChannel channel) {
404         final Thread currentThread = Thread.currentThread();
405         final Thread workerThread = thread;
406         if (currentThread != workerThread) {
407             if (channel.writeTaskInTaskQueue.compareAndSet(false, true)) {
408                 boolean offered = writeTaskQueue.offer(channel.writeTask);
409                 assert offered;
410             }
411 
412             if (!(channel instanceof NioAcceptedSocketChannel) ||
413                 ((NioAcceptedSocketChannel) channel).bossThread != currentThread) {
414                 final Selector workerSelector = selector;
415                 if (workerSelector != null) {
416                     if (wakenUp.compareAndSet(false, true)) {
417                         workerSelector.wakeup();
418                     }
419                 }
420             } else {
421                 // A write request can be made from an acceptor thread (boss)
422                 // when a user attempted to write something in:
423                 //
424                 //   * channelOpen()
425                 //   * channelBound()
426                 //   * channelConnected().
427                 //
428                 // In this case, there's no need to wake up the selector because
429                 // the channel is not even registered yet at this moment.
430             }
431 
432             return true;
433         }
434 
435         return false;
436     }
437 
438     private void write0(NioSocketChannel channel) {
439         boolean open = true;
440         boolean addOpWrite = false;
441         boolean removeOpWrite = false;
442 
443         long writtenBytes = 0;
444 
445         final SocketSendBufferPool sendBufferPool = this.sendBufferPool;
446         final SocketChannel ch = channel.socket;
447         final Queue<MessageEvent> writeBuffer = channel.writeBuffer;
448         final int writeSpinCount = channel.getConfig().getWriteSpinCount();
449         synchronized (channel.writeLock) {
450             channel.inWriteNowLoop = true;
451             for (;;) {
452                 MessageEvent evt = channel.currentWriteEvent;
453                 SendBuffer buf;
454                 if (evt == null) {
455                     if ((channel.currentWriteEvent = evt = writeBuffer.poll()) == null) {
456                         removeOpWrite = true;
457                         channel.writeSuspended = false;
458                         break;
459                     }
460 
461                     channel.currentWriteBuffer = buf = sendBufferPool.acquire(evt.getMessage());
462                 } else {
463                     buf = channel.currentWriteBuffer;
464                 }
465 
466                 ChannelFuture future = evt.getFuture();
467                 try {
468                     long localWrittenBytes = 0;
469                     for (int i = writeSpinCount; i > 0; i --) {
470                         localWrittenBytes = buf.transferTo(ch);
471                         if (localWrittenBytes != 0) {
472                             writtenBytes += localWrittenBytes;
473                             break;
474                         }
475                         if (buf.finished()) {
476                             break;
477                         }
478                     }
479 
480                     if (buf.finished()) {
481                         // Successful write - proceed to the next message.
482                         buf.release();
483                         channel.currentWriteEvent = null;
484                         channel.currentWriteBuffer = null;
485                         evt = null;
486                         buf = null;
487                         future.setSuccess();
488                     } else {
489                         // Not written fully - perhaps the kernel buffer is full.
490                         addOpWrite = true;
491                         channel.writeSuspended = true;
492 
493                         if (localWrittenBytes > 0) {
494                             // Notify progress listeners if necessary.
495                             future.setProgress(
496                                     localWrittenBytes,
497                                     buf.writtenBytes(), buf.totalBytes());
498                         }
499                         break;
500                     }
501                 } catch (AsynchronousCloseException e) {
502                     // Doesn't need a user attention - ignore.
503                 } catch (Throwable t) {
504                     buf.release();
505                     channel.currentWriteEvent = null;
506                     channel.currentWriteBuffer = null;
507                     buf = null;
508                     evt = null;
509                     future.setFailure(t);
510                     fireExceptionCaught(channel, t);
511                     if (t instanceof IOException) {
512                         open = false;
513                         close(channel, succeededFuture(channel));
514                     }
515                 }
516             }
517             channel.inWriteNowLoop = false;
518         }
519 
520         fireWriteComplete(channel, writtenBytes);
521 
522         if (open) {
523             if (addOpWrite) {
524                 setOpWrite(channel);
525             } else if (removeOpWrite) {
526                 clearOpWrite(channel);
527             }
528         }
529     }
530 
531     private void setOpWrite(NioSocketChannel channel) {
532         Selector selector = this.selector;
533         SelectionKey key = channel.socket.keyFor(selector);
534         if (key == null) {
535             return;
536         }
537         if (!key.isValid()) {
538             close(key);
539             return;
540         }
541 
542         // interestOps can change at any time and at any thread.
543         // Acquire a lock to avoid possible race condition.
544         synchronized (channel.interestOpsLock) {
545             int interestOps = channel.getRawInterestOps();
546             if ((interestOps & SelectionKey.OP_WRITE) == 0) {
547                 interestOps |= SelectionKey.OP_WRITE;
548                 key.interestOps(interestOps);
549                 channel.setRawInterestOpsNow(interestOps);
550             }
551         }
552     }
553 
554     private void clearOpWrite(NioSocketChannel channel) {
555         Selector selector = this.selector;
556         SelectionKey key = channel.socket.keyFor(selector);
557         if (key == null) {
558             return;
559         }
560         if (!key.isValid()) {
561             close(key);
562             return;
563         }
564 
565         // interestOps can change at any time and at any thread.
566         // Acquire a lock to avoid possible race condition.
567         synchronized (channel.interestOpsLock) {
568             int interestOps = channel.getRawInterestOps();
569             if ((interestOps & SelectionKey.OP_WRITE) != 0) {
570                 interestOps &= ~SelectionKey.OP_WRITE;
571                 key.interestOps(interestOps);
572                 channel.setRawInterestOpsNow(interestOps);
573             }
574         }
575     }
576 
577     void close(NioSocketChannel channel, ChannelFuture future) {
578         boolean connected = channel.isConnected();
579         boolean bound = channel.isBound();
580         try {
581             channel.socket.close();
582             cancelledKeys ++;
583 
584             if (channel.setClosed()) {
585                 future.setSuccess();
586                 if (connected) {
587                     fireChannelDisconnected(channel);
588                 }
589                 if (bound) {
590                     fireChannelUnbound(channel);
591                 }
592 
593                 cleanUpWriteBuffer(channel);
594                 fireChannelClosed(channel);
595             } else {
596                 future.setSuccess();
597             }
598         } catch (Throwable t) {
599             future.setFailure(t);
600             fireExceptionCaught(channel, t);
601         }
602     }
603 
604     private void cleanUpWriteBuffer(NioSocketChannel channel) {
605         Exception cause = null;
606         boolean fireExceptionCaught = false;
607 
608         // Clean up the stale messages in the write buffer.
609         synchronized (channel.writeLock) {
610             MessageEvent evt = channel.currentWriteEvent;
611             if (evt != null) {
612                 // Create the exception only once to avoid the excessive overhead
613                 // caused by fillStackTrace.
614                 if (channel.isOpen()) {
615                     cause = new NotYetConnectedException();
616                 } else {
617                     cause = new ClosedChannelException();
618                 }
619 
620                 ChannelFuture future = evt.getFuture();
621                 channel.currentWriteBuffer.release();
622                 channel.currentWriteBuffer = null;
623                 channel.currentWriteEvent = null;
624                 evt = null;
625                 future.setFailure(cause);
626                 fireExceptionCaught = true;
627             }
628 
629             Queue<MessageEvent> writeBuffer = channel.writeBuffer;
630             if (!writeBuffer.isEmpty()) {
631                 // Create the exception only once to avoid the excessive overhead
632                 // caused by fillStackTrace.
633                 if (cause == null) {
634                     if (channel.isOpen()) {
635                         cause = new NotYetConnectedException();
636                     } else {
637                         cause = new ClosedChannelException();
638                     }
639                 }
640 
641                 for (;;) {
642                     evt = writeBuffer.poll();
643                     if (evt == null) {
644                         break;
645                     }
646                     evt.getFuture().setFailure(cause);
647                     fireExceptionCaught = true;
648                 }
649             }
650         }
651 
652         if (fireExceptionCaught) {
653             fireExceptionCaught(channel, cause);
654         }
655     }
656 
657     void setInterestOps(
658             NioSocketChannel channel, ChannelFuture future, int interestOps) {
659         boolean changed = false;
660         try {
661             // interestOps can change at any time and at any thread.
662             // Acquire a lock to avoid possible race condition.
663             synchronized (channel.interestOpsLock) {
664                 Selector selector = this.selector;
665                 SelectionKey key = channel.socket.keyFor(selector);
666 
667                 if (key == null || selector == null) {
668                     // Not registered to the worker yet.
669                     // Set the rawInterestOps immediately; RegisterTask will pick it up.
670                     channel.setRawInterestOpsNow(interestOps);
671                     return;
672                 }
673 
674                 // Override OP_WRITE flag - a user cannot change this flag.
675                 interestOps &= ~Channel.OP_WRITE;
676                 interestOps |= channel.getRawInterestOps() & Channel.OP_WRITE;
677 
678                 switch (CONSTRAINT_LEVEL) {
679                 case 0:
680                     if (channel.getRawInterestOps() != interestOps) {
681                         key.interestOps(interestOps);
682                         if (Thread.currentThread() != thread &&
683                             wakenUp.compareAndSet(false, true)) {
684                             selector.wakeup();
685                         }
686                         changed = true;
687                     }
688                     break;
689                 case 1:
690                 case 2:
691                     if (channel.getRawInterestOps() != interestOps) {
692                         if (Thread.currentThread() == thread) {
693                             key.interestOps(interestOps);
694                             changed = true;
695                         } else {
696                             selectorGuard.readLock().lock();
697                             try {
698                                 if (wakenUp.compareAndSet(false, true)) {
699                                     selector.wakeup();
700                                 }
701                                 key.interestOps(interestOps);
702                                 changed = true;
703                             } finally {
704                                 selectorGuard.readLock().unlock();
705                             }
706                         }
707                     }
708                     break;
709                 default:
710                     throw new Error();
711                 }
712 
713                 if (changed) {
714                     channel.setRawInterestOpsNow(interestOps);
715                 }
716             }
717 
718             future.setSuccess();
719             if (changed) {
720                 fireChannelInterestChanged(channel);
721             }
722         } catch (CancelledKeyException e) {
723             // setInterestOps() was called on a closed channel.
724             ClosedChannelException cce = new ClosedChannelException();
725             future.setFailure(cce);
726             fireExceptionCaught(channel, cce);
727         } catch (Throwable t) {
728             future.setFailure(t);
729             fireExceptionCaught(channel, t);
730         }
731     }
732 
733     private final class RegisterTask implements Runnable {
734         private final NioSocketChannel channel;
735         private final ChannelFuture future;
736         private final boolean server;
737 
738         RegisterTask(
739                 NioSocketChannel channel, ChannelFuture future, boolean server) {
740 
741             this.channel = channel;
742             this.future = future;
743             this.server = server;
744         }
745 
746         public void run() {
747             SocketAddress localAddress = channel.getLocalAddress();
748             SocketAddress remoteAddress = channel.getRemoteAddress();
749             if (localAddress == null || remoteAddress == null) {
750                 if (future != null) {
751                     future.setFailure(new ClosedChannelException());
752                 }
753                 close(channel, succeededFuture(channel));
754                 return;
755             }
756 
757             try {
758                 if (server) {
759                     channel.socket.configureBlocking(false);
760                 }
761 
762                 synchronized (channel.interestOpsLock) {
763                     channel.socket.register(
764                             selector, channel.getRawInterestOps(), channel);
765                 }
766                 if (future != null) {
767                     channel.setConnected();
768                     future.setSuccess();
769                 }
770             } catch (IOException e) {
771                 if (future != null) {
772                     future.setFailure(e);
773                 }
774                 close(channel, succeededFuture(channel));
775                 if (!(e instanceof ClosedChannelException)) {
776                     throw new ChannelException(
777                             "Failed to register a socket to the selector.", e);
778                 }
779             }
780 
781             if (!server) {
782                 if (!((NioClientSocketChannel) channel).boundManually) {
783                     fireChannelBound(channel, localAddress);
784                 }
785                 fireChannelConnected(channel, remoteAddress);
786             }
787         }
788     }
789 }