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.DatagramChannel;
    27  import java.nio.channels.NotYetConnectedException;
    28  import java.nio.channels.SelectionKey;
    29  import java.nio.channels.Selector;
    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.ChannelBufferFactory;
    40  import org.jboss.netty.channel.Channel;
    41  import org.jboss.netty.channel.ChannelException;
    42  import org.jboss.netty.channel.ChannelFuture;
    43  import org.jboss.netty.channel.MessageEvent;
    44  import org.jboss.netty.channel.ReceiveBufferSizePredictor;
    45  import org.jboss.netty.channel.socket.nio.SocketSendBufferPool.SendBuffer;
    46  import org.jboss.netty.logging.InternalLogger;
    47  import org.jboss.netty.logging.InternalLoggerFactory;
    48  import org.jboss.netty.util.ThreadRenamingRunnable;
    49  import org.jboss.netty.util.internal.LinkedTransferQueue;
    50
    51  /**
    52   * A class responsible for registering channels with {@link Selector}.
    53   * It also implements the {@link Selector} loop.
    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   * @author Daniel Bevenius (dbevenius@jboss.com)
    58   *
    59   * @version $Rev: 2376 $, $Date: 2010-10-25 03:24:20 +0900 (Mon, 25 Oct 2010) $
    60   */
    61  class NioDatagramWorker implements Runnable {
    62      /**
    63       * Internal Netty logger.
    64       */
    65      private static final InternalLogger logger = InternalLoggerFactory
    66              .getInstance(NioDatagramWorker.class);
    67
    68      /**
    69       * This id of this worker.
    70       */
    71      private final int id;
    72
    73      /**
    74       * This id of the NioDatagramPipelineSink.
    75       */
    76      private final int bossId;
    77
    78      /**
    79       * Executor used to execute {@link Runnable}s such as
    80       * {@link ChannelRegistionTask}.
    81       */
    82      private final Executor executor;
    83
    84      /**
    85       * Boolean to indicate if this worker has been started.
    86       */
    87      private boolean started;
    88
    89      /**
    90       * If this worker has been started thread will be a reference to the thread
    91       * used when starting. i.e. the current thread when the run method is executed.
    92       */
    93      private volatile Thread thread;
    94
    95      /**
    96       * The NIO {@link Selector}.
    97       */
    98      volatile Selector selector;
    99
    100     /**
    101      * Boolean that controls determines if a blocked Selector.select should
    102      * break out of its selection process. In our case we use a timeone for
    103      * the select method and the select method will block for that time unless
    104      * waken up.
    105      */
    106     private final AtomicBoolean wakenUp = new AtomicBoolean();
    107
    108     /**
    109      * Lock for this workers Selector.
    110      */
    111     private final ReadWriteLock selectorGuard = new ReentrantReadWriteLock();
    112
    113     /**
    114      * Monitor object used to synchronize selector open/close.
    115      */
    116     private final Object startStopLock = new Object();
    117
    118     /**
    119      * Queue of {@link ChannelRegistionTask}s
    120      */
    121     private final Queue<Runnable> registerTaskQueue = new LinkedTransferQueue<Runnable>();
    122
    123     /**
    124      * Queue of WriteTasks
    125      */
    126     private final Queue<Runnable> writeTaskQueue = new LinkedTransferQueue<Runnable>();
    127
    128     private volatile int cancelledKeys; // should use AtomicInteger but we just need approximation
    129
    130     private final SocketSendBufferPool sendBufferPool = new SocketSendBufferPool();
    131
    132     /**
    133      * Sole constructor.
    134      *
    135      * @param bossId This id of the NioDatagramPipelineSink
    136      * @param id The id of this worker
    137      * @param executor the {@link Executor} used to execute {@link Runnable}s
    138      *                 such as {@link ChannelRegistionTask}
    139      */
    140     NioDatagramWorker(final int bossId, final int id, final Executor executor) {
    141         this.bossId = bossId;
    142         this.id = id;
    143         this.executor = executor;
    144     }
    145
    146     /**
    147      * Registers the passed-in channel with a selector.
    148      *
    149      * @param channel the channel to register
    150      * @param future  the {@link ChannelFuture} that has to be notified on
    151      *                completion
    152      */
    153     void register(final NioDatagramChannel channel, final ChannelFuture future) {
    154         final Runnable channelRegTask = new ChannelRegistionTask(channel,
    155                 future);
    156         Selector selector;
    157
    158         synchronized (startStopLock) {
    159             if (!started) {
    160                 // Open a selector if this worker didn't start yet.
    161                 try {
    162                     this.selector = selector = Selector.open();
    163                 } catch (final Throwable t) {
    164                     throw new ChannelException("Failed to create a selector.",
    165                             t);
    166                 }
    167
    168                 boolean success = false;
    169                 try {
    170                     // Start the main selector loop. See run() for details.
    171                     executor.execute(new ThreadRenamingRunnable(this,
    172                             "New I/O datagram worker #" + bossId + "'-'" + id));
    173                     success = true;
    174                 } finally {
    175                     if (!success) {
    176                         try {
    177                             // Release the Selector if the execution fails.
    178                             selector.close();
    179                         } catch (final Throwable t) {
    180                             logger.warn("Failed to close a selector.", t);
    181                         }
    182                         this.selector = selector = null;
    183                         // The method will return to the caller at this point.
    184                     }
    185                 }
    186             } else {
    187                 // Use the existing selector if this worker has been started.
    188                 selector = this.selector;
    189             }
    190             assert selector != null && selector.isOpen();
    191
    192             started = true;
    193
    194             // "Add" the registration task to the register task queue.
    195             boolean offered = registerTaskQueue.offer(channelRegTask);
    196             assert offered;
    197         }
    198
    199         if (wakenUp.compareAndSet(false, true)) {
    200             selector.wakeup();
    201         }
    202     }
    203
    204     /**
    205      * Selector loop.
    206      */
    207     public void run() {
    208         // Store a ref to the current thread.
    209         thread = Thread.currentThread();
    210
    211         final Selector selector = this.selector;
    212         boolean shutdown = false;
    213
    214         for (;;) {
    215             wakenUp.set(false);
    216
    217             if (NioProviderMetadata.CONSTRAINT_LEVEL != 0) {
    218                 selectorGuard.writeLock().lock();
    219                 // This empty synchronization block prevents the selector from acquiring its lock.
    220                 selectorGuard.writeLock().unlock();
    221             }
    222
    223             try {
    224                 SelectorUtil.select(selector);
    225
    226                 // 'wakenUp.compareAndSet(false, true)' is always evaluated
    227                 // before calling 'selector.wakeup()' to reduce the wake-up
    228                 // overhead. (Selector.wakeup() is an expensive operation.)
    229                 //
    230                 // However, there is a race condition in this approach.
    231                 // The race condition is triggered when 'wakenUp' is set to
    232                 // true too early.
    233                 //
    234                 // 'wakenUp' is set to true too early if:
    235                 // 1) Selector is waken up between 'wakenUp.set(false)' and
    236                 //    'selector.select(...)'. (BAD)
    237                 // 2) Selector is waken up between 'selector.select(...)' and
    238                 //    'if (wakenUp.get()) { ... }'. (OK)
    239                 //
    240                 // In the first case, 'wakenUp' is set to true and the
    241                 // following 'selector.select(...)' will wake up immediately.
    242                 // Until 'wakenUp' is set to false again in the next round,
    243                 // 'wakenUp.compareAndSet(false, true)' will fail, and therefore
    244                 // any attempt to wake up the Selector will fail, too, causing
    245                 // the following 'selector.select(...)' call to block
    246                 // unnecessarily.
    247                 //
    248                 // To fix this problem, we wake up the selector again if wakenUp
    249                 // is true immediately after selector.select(...).
    250                 // It is inefficient in that it wakes up the selector for both
    251                 // the first case (BAD - wake-up required) and the second case
    252                 // (OK - no wake-up required).
    253
    254                 if (wakenUp.get()) {
    255                     selector.wakeup();
    256                 }
    257
    258                 cancelledKeys = 0;
    259                 processRegisterTaskQueue();
    260                 processWriteTaskQueue();
    261                 processSelectedKeys(selector.selectedKeys());
    262
    263                 // Exit the loop when there's nothing to handle (the registered
    264                 // key set is empty.
    265                 // The shutdown flag is used to delay the shutdown of this
    266                 // loop to avoid excessive Selector creation when
    267                 // connections are registered in a one-by-one manner instead of
    268                 // concurrent manner.
    269                 if (selector.keys().isEmpty()) {
    270                     if (shutdown || executor instanceof ExecutorService &&
    271                             ((ExecutorService) executor).isShutdown()) {
    272                         synchronized (startStopLock) {
    273                             if (registerTaskQueue.isEmpty() &&
    274                                     selector.keys().isEmpty()) {
    275                                 started = false;
    276                                 try {
    277                                     selector.close();
    278                                 } catch (IOException e) {
    279                                     logger.warn("Failed to close a selector.",
    280                                             e);
    281                                 } finally {
    282                                     this.selector = null;
    283                                 }
    284                                 break;
    285                             } else {
    286                                 shutdown = false;
    287                             }
    288                         }
    289                     } else {
    290                         // Give one more second.
    291                         shutdown = true;
    292                     }
    293                 } else {
    294                     shutdown = false;
    295                 }
    296             } catch (Throwable t) {
    297                 logger.warn("Unexpected exception in the selector loop.", t);
    298
    299                 // Prevent possible consecutive immediate failures that lead to
    300                 // excessive CPU consumption.
    301                 try {
    302                     Thread.sleep(1000);
    303                 } catch (InterruptedException e) {
    304                     // Ignore.
    305                 }
    306             }
    307         }
    308     }
    309
    310     /**
    311      * Will go through all the {@link ChannelRegistionTask}s in the
    312      * task queue and run them (registering them).
    313      */
    314     private void processRegisterTaskQueue() throws IOException {
    315         for (;;) {
    316             final Runnable task = registerTaskQueue.poll();
    317             if (task == null) {
    318                 break;
    319             }
    320
    321             task.run();
    322             cleanUpCancelledKeys();
    323         }
    324     }
    325
    326     /**
    327      * Will go through all the WriteTasks and run them.
    328      */
    329     private void processWriteTaskQueue() throws IOException {
    330         for (;;) {
    331             final Runnable task = writeTaskQueue.poll();
    332             if (task == null) {
    333                 break;
    334             }
    335
    336             task.run();
    337             cleanUpCancelledKeys();
    338         }
    339     }
    340
    341     private void processSelectedKeys(final Set<SelectionKey> selectedKeys) throws IOException {
    342         for (Iterator<SelectionKey> i = selectedKeys.iterator(); i.hasNext();) {
    343             SelectionKey k = i.next();
    344             i.remove();
    345             try {
    346                 int readyOps = k.readyOps();
    347                 if ((readyOps & SelectionKey.OP_READ) != 0 || readyOps == 0) {
    348                     if (!read(k)) {
    349                         // Connection already closed - no need to handle write.
    350                         continue;
    351                     }
    352                 }
    353                 if ((readyOps & SelectionKey.OP_WRITE) != 0) {
    354                     writeFromSelectorLoop(k);
    355                 }
    356             } catch (CancelledKeyException e) {
    357                 close(k);
    358             }
    359
    360             if (cleanUpCancelledKeys()) {
    361                 break; // Break the loop to avoid ConcurrentModificationException
    362             }
    363         }
    364     }
    365
    366     private boolean cleanUpCancelledKeys() throws IOException {
    367         if (cancelledKeys >= NioWorker.CLEANUP_INTERVAL) {
    368             cancelledKeys = 0;
    369             selector.selectNow();
    370             return true;
    371         }
    372         return false;
    373     }
    374
    375     /**
    376      * Read is called when a Selector has been notified that the underlying channel
    377      * was something to be read. The channel would previously have registered its interest
    378      * in read operations.
    379      *
    380      * @param key The selection key which contains the Selector registration information.
    381      */
    382     private boolean read(final SelectionKey key) {
    383         final NioDatagramChannel channel = (NioDatagramChannel) key.attachment();
    384         ReceiveBufferSizePredictor predictor =
    385             channel.getConfig().getReceiveBufferSizePredictor();
    386         final ChannelBufferFactory bufferFactory = channel.getConfig().getBufferFactory();
    387         final DatagramChannel nioChannel = (DatagramChannel) key.channel();
    388
    389         // Allocating a non-direct buffer with a max udp packge size.
    390         // Would using a direct buffer be more efficient or would this negatively
    391         // effect performance, as direct buffer allocation has a higher upfront cost
    392         // where as a ByteBuffer is heap allocated.
    393         final ByteBuffer byteBuffer = ByteBuffer.allocate(
    394                 predictor.nextReceiveBufferSize()).order(bufferFactory.getDefaultOrder());
    395
    396         boolean failure = true;
    397         SocketAddress remoteAddress = null;
    398         try {
    399             // Receive from the channel in a non blocking mode. We have already been notified that
    400             // the channel is ready to receive.
    401             remoteAddress = nioChannel.receive(byteBuffer);
    402             failure = false;
    403         } catch (ClosedChannelException e) {
    404             // Can happen, and does not need a user attention.
    405         } catch (Throwable t) {
    406             fireExceptionCaught(channel, t);
    407         }
    408
    409         if (remoteAddress != null) {
    410             // Flip the buffer so that we can wrap it.
    411             byteBuffer.flip();
    412
    413             int readBytes = byteBuffer.remaining();
    414             if (readBytes > 0) {
    415                 // Update the predictor.
    416                 predictor.previousReceiveBufferSize(readBytes);
    417
    418                 // Notify the interested parties about the newly arrived message.
    419                 fireMessageReceived(
    420                         channel, bufferFactory.getBuffer(byteBuffer), remoteAddress);
    421             }
    422         }
    423
    424         if (failure) {
    425             close(channel, succeededFuture(channel));
    426             return false;
    427         }
    428
    429         return true;
    430     }
    431
    432     private void close(SelectionKey k) {
    433         final NioDatagramChannel ch = (NioDatagramChannel) k.attachment();
    434         close(ch, succeededFuture(ch));
    435     }
    436
    437     void writeFromUserCode(final NioDatagramChannel channel) {
    438         /*
    439          * Note that we are not checking if the channel is connected. Connected
    440          * has a different meaning in UDP and means that the channels socket is
    441          * configured to only send and receive from a given remote peer.
    442          */
    443         if (!channel.isOpen()) {
    444             cleanUpWriteBuffer(channel);
    445             return;
    446         }
    447
    448         if (scheduleWriteIfNecessary(channel)) {
    449             return;
    450         }
    451
    452         // From here, we are sure Thread.currentThread() == workerThread.
    453
    454         if (channel.writeSuspended) {
    455             return;
    456         }
    457
    458         if (channel.inWriteNowLoop) {
    459             return;
    460         }
    461
    462         write0(channel);
    463     }
    464
    465     void writeFromTaskLoop(final NioDatagramChannel ch) {
    466         if (!ch.writeSuspended) {
    467             write0(ch);
    468         }
    469     }
    470
    471     void writeFromSelectorLoop(final SelectionKey k) {
    472         NioDatagramChannel ch = (NioDatagramChannel) k.attachment();
    473         ch.writeSuspended = false;
    474         write0(ch);
    475     }
    476
    477     private boolean scheduleWriteIfNecessary(final NioDatagramChannel channel) {
    478         final Thread workerThread = thread;
    479         if (workerThread == null || Thread.currentThread() != workerThread) {
    480             if (channel.writeTaskInTaskQueue.compareAndSet(false, true)) {
    481                 // "add" the channels writeTask to the writeTaskQueue.
    482                 boolean offered = writeTaskQueue.offer(channel.writeTask);
    483                 assert offered;
    484             }
    485
    486             final Selector selector = this.selector;
    487             if (selector != null) {
    488                 if (wakenUp.compareAndSet(false, true)) {
    489                     selector.wakeup();
    490                 }
    491             }
    492             return true;
    493         }
    494
    495         return false;
    496     }
    497
    498     private void write0(final NioDatagramChannel channel) {
    499
    500         boolean addOpWrite = false;
    501         boolean removeOpWrite = false;
    502
    503         long writtenBytes = 0;
    504
    505         final SocketSendBufferPool sendBufferPool = this.sendBufferPool;
    506         final DatagramChannel ch = channel.getDatagramChannel();
    507         final Queue<MessageEvent> writeBuffer = channel.writeBufferQueue;
    508         final int writeSpinCount = channel.getConfig().getWriteSpinCount();
    509         synchronized (channel.writeLock) {
    510             // inform the channel that write is in-progress
    511             channel.inWriteNowLoop = true;
    512
    513             // loop forever...
    514             for (;;) {
    515                 MessageEvent evt = channel.currentWriteEvent;
    516                 SendBuffer buf;
    517                 if (evt == null) {
    518                     if ((channel.currentWriteEvent = evt = writeBuffer.poll()) == null) {
    519                         removeOpWrite = true;
    520                         channel.writeSuspended = false;
    521                         break;
    522                     }
    523
    524                     channel.currentWriteBuffer = buf = sendBufferPool.acquire(evt.getMessage());
    525                 } else {
    526                     buf = channel.currentWriteBuffer;
    527                 }
    528
    529                 try {
    530                     long localWrittenBytes = 0;
    531                     SocketAddress raddr = evt.getRemoteAddress();
    532                     if (raddr == null) {
    533                         for (int i = writeSpinCount; i > 0; i --) {
    534                             localWrittenBytes = buf.transferTo(ch);
    535                             if (localWrittenBytes != 0) {
    536                                 writtenBytes += localWrittenBytes;
    537                                 break;
    538                             }
    539                             if (buf.finished()) {
    540                                 break;
    541                             }
    542                         }
    543                     } else {
    544                         for (int i = writeSpinCount; i > 0; i --) {
    545                             localWrittenBytes = buf.transferTo(ch, raddr);
    546                             if (localWrittenBytes != 0) {
    547                                 writtenBytes += localWrittenBytes;
    548                                 break;
    549                             }
    550                             if (buf.finished()) {
    551                                 break;
    552                             }
    553                         }
    554                     }
    555
    556                     if (localWrittenBytes > 0 || buf.finished()) {
    557                         // Successful write - proceed to the next message.
    558                         buf.release();
    559                         ChannelFuture future = evt.getFuture();
    560                         channel.currentWriteEvent = null;
    561                         channel.currentWriteBuffer = null;
    562                         evt = null;
    563                         buf = null;
    564                         future.setSuccess();
    565                     } else {
    566                         // Not written at all - perhaps the kernel buffer is full.
    567                         addOpWrite = true;
    568                         channel.writeSuspended = true;
    569                         break;
    570                     }
    571                 } catch (final AsynchronousCloseException e) {
    572                     // Doesn't need a user attention - ignore.
    573                 } catch (final Throwable t) {
    574                     buf.release();
    575                     ChannelFuture future = evt.getFuture();
    576                     channel.currentWriteEvent = null;
    577                     channel.currentWriteBuffer = null;
    578                     buf = null;
    579                     evt = null;
    580                     future.setFailure(t);
    581                     fireExceptionCaught(channel, t);
    582                 }
    583             }
    584             channel.inWriteNowLoop = false;
    585         }
    586
    587         fireWriteComplete(channel, writtenBytes);
    588
    589         if (addOpWrite) {
    590             setOpWrite(channel);
    591         } else if (removeOpWrite) {
    592             clearOpWrite(channel);
    593         }
    594     }
    595
    596     private void setOpWrite(final NioDatagramChannel channel) {
    597         Selector selector = this.selector;
    598         SelectionKey key = channel.getDatagramChannel().keyFor(selector);
    599         if (key == null) {
    600             return;
    601         }
    602         if (!key.isValid()) {
    603             close(key);
    604             return;
    605         }
    606
    607         // interestOps can change at any time and at any thread.
    608         // Acquire a lock to avoid possible race condition.
    609         synchronized (channel.interestOpsLock) {
    610             int interestOps = channel.getRawInterestOps();
    611             if ((interestOps & SelectionKey.OP_WRITE) == 0) {
    612                 interestOps |= SelectionKey.OP_WRITE;
    613                 key.interestOps(interestOps);
    614                 channel.setRawInterestOpsNow(interestOps);
    615             }
    616         }
    617     }
    618
    619     private void clearOpWrite(NioDatagramChannel channel) {
    620         Selector selector = this.selector;
    621         SelectionKey key = channel.getDatagramChannel().keyFor(selector);
    622         if (key == null) {
    623             return;
    624         }
    625         if (!key.isValid()) {
    626             close(key);
    627             return;
    628         }
    629
    630         // interestOps can change at any time and at any thread.
    631         // Acquire a lock to avoid possible race condition.
    632         synchronized (channel.interestOpsLock) {
    633             int interestOps = channel.getRawInterestOps();
    634             if ((interestOps & SelectionKey.OP_WRITE) != 0) {
    635                 interestOps &= ~SelectionKey.OP_WRITE;
    636                 key.interestOps(interestOps);
    637                 channel.setRawInterestOpsNow(interestOps);
    638             }
    639         }
    640     }
    641
    642     static void disconnect(NioDatagramChannel channel, ChannelFuture future) {
    643         boolean connected = channel.isConnected();
    644         try {
    645             channel.getDatagramChannel().disconnect();
    646             future.setSuccess();
    647             if (connected) {
    648                 fireChannelDisconnected(channel);
    649             }
    650         } catch (Throwable t) {
    651             future.setFailure(t);
    652             fireExceptionCaught(channel, t);
    653         }
    654     }
    655
    656     void close(final NioDatagramChannel channel,
    657             final ChannelFuture future) {
    658         boolean connected = channel.isConnected();
    659         boolean bound = channel.isBound();
    660         try {
    661             channel.getDatagramChannel().close();
    662             cancelledKeys ++;
    663
    664             if (channel.setClosed()) {
    665                 future.setSuccess();
    666                 if (connected) {
    667                     fireChannelDisconnected(channel);
    668                 }
    669                 if (bound) {
    670                     fireChannelUnbound(channel);
    671                 }
    672
    673                 cleanUpWriteBuffer(channel);
    674                 fireChannelClosed(channel);
    675             } else {
    676                 future.setSuccess();
    677             }
    678         } catch (Throwable t) {
    679             future.setFailure(t);
    680             fireExceptionCaught(channel, t);
    681         }
    682     }
    683
    684     private void cleanUpWriteBuffer(final NioDatagramChannel channel) {
    685         Exception cause = null;
    686         boolean fireExceptionCaught = false;
    687
    688         // Clean up the stale messages in the write buffer.
    689         synchronized (channel.writeLock) {
    690             MessageEvent evt = channel.currentWriteEvent;
    691             if (evt != null) {
    692                 // Create the exception only once to avoid the excessive overhead
    693                 // caused by fillStackTrace.
    694                 if (channel.isOpen()) {
    695                     cause = new NotYetConnectedException();
    696                 } else {
    697                     cause = new ClosedChannelException();
    698                 }
    699
    700                 ChannelFuture future = evt.getFuture();
    701                 channel.currentWriteBuffer.release();
    702                 channel.currentWriteBuffer = null;
    703                 channel.currentWriteEvent = null;
    704                 evt = null;
    705                 future.setFailure(cause);
    706                 fireExceptionCaught = true;
    707             }
    708
    709             Queue<MessageEvent> writeBuffer = channel.writeBufferQueue;
    710             if (!writeBuffer.isEmpty()) {
    711                 // Create the exception only once to avoid the excessive overhead
    712                 // caused by fillStackTrace.
    713                 if (cause == null) {
    714                     if (channel.isOpen()) {
    715                         cause = new NotYetConnectedException();
    716                     } else {
    717                         cause = new ClosedChannelException();
    718                     }
    719                 }
    720
    721                 for (;;) {
    722                     evt = writeBuffer.poll();
    723                     if (evt == null) {
    724                         break;
    725                     }
    726                     evt.getFuture().setFailure(cause);
    727                     fireExceptionCaught = true;
    728                 }
    729             }
    730         }
    731
    732         if (fireExceptionCaught) {
    733             fireExceptionCaught(channel, cause);
    734         }
    735     }
    736
    737     void setInterestOps(final NioDatagramChannel channel,
    738             ChannelFuture future, int interestOps) {
    739
    740         boolean changed = false;
    741         try {
    742             // interestOps can change at any time and by any thread.
    743             // Acquire a lock to avoid possible race condition.
    744             synchronized (channel.interestOpsLock) {
    745                 final Selector selector = this.selector;
    746                 final SelectionKey key = channel.getDatagramChannel().keyFor(selector);
    747
    748                 if (key == null || selector == null) {
    749                     // Not registered to the worker yet.
    750                     // Set the rawInterestOps immediately; RegisterTask will pick it up.
    751                     channel.setRawInterestOpsNow(interestOps);
    752                     return;
    753                 }
    754
    755                 // Override OP_WRITE flag - a user cannot change this flag.
    756                 interestOps &= ~Channel.OP_WRITE;
    757                 interestOps |= channel.getRawInterestOps() & Channel.OP_WRITE;
    758
    759                 switch (NioProviderMetadata.CONSTRAINT_LEVEL) {
    760                 case 0:
    761                     if (channel.getRawInterestOps() != interestOps) {
    762                         // Set the interesteOps on the SelectionKey
    763                         key.interestOps(interestOps);
    764                         // If the worker thread (the one that that might possibly be blocked
    765                         // in a select() call) is not the thread executing this method wakeup
    766                         // the select() operation.
    767                         if (Thread.currentThread() != thread &&
    768                                 wakenUp.compareAndSet(false, true)) {
    769                             selector.wakeup();
    770                         }
    771                         changed = true;
    772                     }
    773                     break;
    774                 case 1:
    775                 case 2:
    776                     if (channel.getRawInterestOps() != interestOps) {
    777                         if (Thread.currentThread() == thread) {
    778                             // Going to set the interestOps from the same thread.
    779                             // Set the interesteOps on the SelectionKey
    780                             key.interestOps(interestOps);
    781                             changed = true;
    782                         } else {
    783                             // Going to set the interestOps from a different thread
    784                             // and some old provides will need synchronization.
    785                             selectorGuard.readLock().lock();
    786                             try {
    787                                 if (wakenUp.compareAndSet(false, true)) {
    788                                     selector.wakeup();
    789                                 }
    790                                 key.interestOps(interestOps);
    791                                 changed = true;
    792                             } finally {
    793                                 selectorGuard.readLock().unlock();
    794                             }
    795                         }
    796                     }
    797                     break;
    798                 default:
    799                     throw new Error();
    800                 }
    801                 if (changed) {
    802                     channel.setRawInterestOpsNow(interestOps);
    803                 }
    804             }
    805
    806             future.setSuccess();
    807             if (changed) {
    808                 fireChannelInterestChanged(channel);
    809             }
    810         } catch (final CancelledKeyException e) {
    811             // setInterestOps() was called on a closed channel.
    812             ClosedChannelException cce = new ClosedChannelException();
    813             future.setFailure(cce);
    814             fireExceptionCaught(channel, cce);
    815         } catch (final Throwable t) {
    816             future.setFailure(t);
    817             fireExceptionCaught(channel, t);
    818         }
    819     }
    820
    821     /**
    822      * RegisterTask is a task responsible for registering a channel with a
    823      * selector.
    824      */
    825     private final class ChannelRegistionTask implements Runnable {
    826         private final NioDatagramChannel channel;
    827
    828         private final ChannelFuture future;
    829
    830         ChannelRegistionTask(final NioDatagramChannel channel,
    831                 final ChannelFuture future) {
    832             this.channel = channel;
    833             this.future = future;
    834         }
    835
    836         /**
    837          * This runnable's task. Does the actual registering by calling the
    838          * underlying DatagramChannels peer DatagramSocket register method.
    839          *
    840          */
    841         public void run() {
    842             final SocketAddress localAddress = channel.getLocalAddress();
    843             if (localAddress == null) {
    844                 if (future != null) {
    845                     future.setFailure(new ClosedChannelException());
    846                 }
    847                 close(channel, succeededFuture(channel));
    848                 return;
    849             }
    850
    851             try {
    852                 synchronized (channel.interestOpsLock) {
    853                     channel.getDatagramChannel().register(
    854                             selector, channel.getRawInterestOps(), channel);
    855                 }
    856                 if (future != null) {
    857                     future.setSuccess();
    858                 }
    859             } catch (final ClosedChannelException e) {
    860                 if (future != null) {
    861                     future.setFailure(e);
    862                 }
    863                 close(channel, succeededFuture(channel));
    864                 throw new ChannelException(
    865                         "Failed to register a socket to the selector.", e);
    866             }
    867         }
    868     }
    869 }