View Javadoc

    1   /***
    2    *
    3    * Copyright 2003-2004 The Apache Software Foundation
    4    *
    5    *  Licensed under the Apache License, Version 2.0 (the "License");
    6    *  you may not use this file except in compliance with the License.
    7    *  You may obtain a copy of the License at
    8    *
    9    *     http://www.apache.org/licenses/LICENSE-2.0
    10   *
    11   *  Unless required by applicable law or agreed to in writing, software
    12   *  distributed under the License is distributed on an "AS IS" BASIS,
    13   *  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
    14   *  See the License for the specific language governing permissions and
    15   *  limitations under the License.
    16   */
    17
    18  package org.activeio.net;
    19
    20  import java.io.IOException;
    21  import java.nio.channels.ClosedChannelException;
    22  import java.nio.channels.SelectionKey;
    23  import java.nio.channels.Selector;
    24  import java.nio.channels.SocketChannel;
    25  import java.util.Iterator;
    26  import java.util.LinkedList;
    27  import java.util.Set;
    28
    29  import org.activeio.ChannelFactory;
    30
    31  import EDU.oswego.cs.dl.util.concurrent.Executor;
    32  import EDU.oswego.cs.dl.util.concurrent.DirectExecutor;
    33
    34  /***
    35   * The SelectorManager will manage one Selector and the thread that checks the
    36   * selector.
    37   *
    38   * We may need to consider running more than one thread to check the selector if
    39   * servicing the selector takes too long.
    40   *
    41   * @version $Rev: 46019 $ $Date: 2004-09-14 05:56:06 -0400 (Tue, 14 Sep 2004) $
    42   */
    43  final public class NIOAsynchChannelSelectorManager {
    44
    45      static private Executor selectorExecutor = ChannelFactory.DEFAULT_EXECUTOR;
    46      static private Executor channelExecutor = ChannelFactory.DEFAULT_EXECUTOR;
    47
    48      static private LinkedList freeManagers = new LinkedList();
    49      static private LinkedList fullManagers = new LinkedList();
    50      private static final int MAX_CHANNELS_PER_SELECTOR  = 50;
    51
    52      static {
    53         String os = System.getProperty("os.name");
    54         if( os.startsWith("Linux") ) {
    55             channelExecutor = new DirectExecutor();
    56         }
    57      }
    58
    59      public static interface SelectorManagerListener {
    60          public void onSelect(SocketChannelAsynchChannelSelection selector);
    61      }
    62
    63      final public class SocketChannelAsynchChannelSelection {
    64
    65          private final SelectionKey key;
    66          private final SelectorManagerListener listener;
    67          private boolean closed;
    68          private int interest;
    69
    70          private SocketChannelAsynchChannelSelection(SocketChannel socketChannel, SelectorManagerListener listener)
    71                  throws ClosedChannelException {
    72              this.listener = listener;
    73              this.key = socketChannel.register(selector, 0, this);
    74              incrementUseCounter();
    75          }
    76
    77          public void setInterestOps(int ops) {
    78              	if( closed )
    79              		return;
    80              	interest = ops;
    81               enable();
    82          }
    83
    84          public void enable() {
    85              if( closed )
    86                  return;
    87              key.interestOps(interest);
    88              selector.wakeup();
    89          }
    90
    91          public void disable() {
    92              if( closed )
    93                  return;
    94              key.interestOps(0);
    95          }
    96
    97          public void close() {
    98          	if( closed )
    99          		return;
    100
    101             key.cancel();
    102             decrementUseCounter();
    103             selector.wakeup();
    104             closed=true;
    105         }
    106
    107         public void onSelect() {
    108             if( !key.isValid() )
    109                 return;
    110             listener.onSelect(this);
    111         }
    112
    113         public boolean isWritable() {
    114             return key.isWritable();
    115         }
    116
    117         public boolean isReadable() {
    118             return key.isReadable();
    119         }
    120     }
    121
    122     public synchronized static SocketChannelAsynchChannelSelection register(
    123             SocketChannel socketChannel, SelectorManagerListener listener)
    124             throws IOException {
    125
    126         NIOAsynchChannelSelectorManager manager = null;
    127         synchronized (freeManagers) {
    128             if (freeManagers.size() > 0)
    129                 manager = (NIOAsynchChannelSelectorManager) freeManagers.getFirst();
    130             if (manager == null) {
    131                 manager = new NIOAsynchChannelSelectorManager();
    132                 freeManagers.addFirst(manager);
    133             }
    134
    135             // That manager may have filled up.
    136             SocketChannelAsynchChannelSelection selection = manager.new SocketChannelAsynchChannelSelection(
    137                     socketChannel, listener);
    138             if (manager.useCounter >= MAX_CHANNELS_PER_SELECTOR) {
    139                 freeManagers.removeFirst();
    140                 fullManagers.addLast(manager);
    141             }
    142             return selection;
    143         }
    144     }
    145
    146     public synchronized static void setSelectorExecutor(Executor executor) {
    147         NIOAsynchChannelSelectorManager.selectorExecutor = executor;
    148     }
    149
    150     public synchronized static void setChannelExecutor(Executor executor) {
    151         NIOAsynchChannelSelectorManager.channelExecutor = executor;
    152     }
    153
    154     private class SelectorWorker implements Runnable {
    155
    156         public void run() {
    157
    158             String origName = Thread.currentThread().getName();
    159             try {
    160                Thread.currentThread().setName("Selector Worker: "+getId());
    161                while ( isRunning() ) {
    162
    163                    int count = selector.select(10);
    164                    if (count == 0)
    165                        continue;
    166                     if( !isRunning() )
    167                         return;
    168
    169                     // Get a java.util.Set containing the SelectionKey objects
    170                     // for all channels that are ready for I/O.
    171                     Set keys = selector.selectedKeys();
    172
    173                     for (Iterator i = keys.iterator(); i.hasNext();) {
    174                         final SelectionKey key = (SelectionKey) i.next();
    175                         i.remove();
    176
    177                         if( !key.isValid() )
    178                             continue;
    179
    180                         final SocketChannelAsynchChannelSelection s = (SocketChannelAsynchChannelSelection) key.attachment();
    181                         s.disable();
    182
    183                         // Kick off another thread to find newly selected keys while we process the
    184                         // currently selected keys
    185                         channelExecutor.execute(new Runnable() {
    186                             public void run() {
    187                                 try {
    188                                     s.onSelect();
    189                                     s.enable();
    190                                 } catch ( Throwable e ) {
    191                                     System.err.println("ActiveIO unexpected error: ");
    192                                     e.printStackTrace(System.err);
    193                                 }
    194                             }
    195                         });
    196                     }
    197
    198                }
    199             } catch (Throwable e) {
    200                 System.err.println("Unexpected exception: " + e);
    201                 e.printStackTrace();
    202             } finally {
    203                 Thread.currentThread().setName(origName);
    204             }
    205         }
    206     }
    207
    208     /***
    209      * The selector used to wait for non-blocking events.
    210      */
    211     private Selector selector;
    212
    213     /***
    214      * How many SelectionKeys does the selector have active.
    215      */
    216     private int useCounter;
    217     private int id = getNextId();
    218     private static int nextId;
    219
    220     private NIOAsynchChannelSelectorManager() throws IOException {
    221         selector = Selector.open();
    222     }
    223
    224     synchronized private static int getNextId() {
    225         return nextId++;
    226     }
    227
    228     private int getId() {
    229         return id ;
    230     }
    231
    232     synchronized private void incrementUseCounter() {
    233         useCounter++;
    234         if (useCounter == 1) {
    235             try {
    236                 selectorExecutor.execute(new SelectorWorker());
    237             } catch (InterruptedException e) {
    238                 Thread.currentThread().interrupt();
    239             }
    240         }
    241     }
    242
    243     synchronized private void decrementUseCounter() {
    244         useCounter--;
    245 	 	synchronized(freeManagers) {
    246  	 		 if( useCounter == 0 ) {
    247   	 		 	freeManagers.remove(this);
    248   	 		 }
    249  	 		 else if( useCounter < MAX_CHANNELS_PER_SELECTOR ) {
    250   	 		 	fullManagers.remove(this);
    251   	 		 	freeManagers.addLast(this);
    252   	 		 }
    253 	    }
    254     }
    255
    256     synchronized private boolean isRunning() {
    257         return useCounter > 0;
    258     }
    259 }