libjava/gnu/java/nio/SelectorImpl.java

   1 /* SelectorImpl.java --
   2    Copyright (C) 2002, 2003, 2004  Free Software Foundation, Inc.
   3
   4 This file is part of GNU Classpath.
   5
   6 GNU Classpath is free software; you can redistribute it and/or modify
   7 it under the terms of the GNU General Public License as published by
   8 the Free Software Foundation; either version 2, or (at your option)
   9 any later version.
  10
  11 GNU Classpath is distributed in the hope that it will be useful, but
  12 WITHOUT ANY WARRANTY; without even the implied warranty of
  13 MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
  14 General Public License for more details.
  15
  16 You should have received a copy of the GNU General Public License
  17 along with GNU Classpath; see the file COPYING.  If not, write to the
  18 Free Software Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA
  19 02111-1307 USA.
  20
  21 Linking this library statically or dynamically with other modules is
  22 making a combined work based on this library.  Thus, the terms and
  23 conditions of the GNU General Public License cover the whole
  24 combination.
  25
  26 As a special exception, the copyright holders of this library give you
  27 permission to link this library with independent modules to produce an
  28 executable, regardless of the license terms of these independent
  29 modules, and to copy and distribute the resulting executable under
  30 terms of your choice, provided that you also meet, for each linked
  31 independent module, the terms and conditions of the license of that
  32 module.  An independent module is a module which is not derived from
  33 or based on this library.  If you modify this library, you may extend
  34 this exception to your version of the library, but you are not
  35 obligated to do so.  If you do not wish to do so, delete this
  36 exception statement from your version. */
  37
  38
  39 package gnu.java.nio;
  40
  41 import java.io.IOException;
  42 import java.nio.channels.ClosedSelectorException;
  43 import java.nio.channels.SelectableChannel;
  44 import java.nio.channels.SelectionKey;
  45 import java.nio.channels.Selector;
  46 import java.nio.channels.spi.AbstractSelectableChannel;
  47 import java.nio.channels.spi.AbstractSelector;
  48 import java.nio.channels.spi.SelectorProvider;
  49 import java.util.Collections;
  50 import java.util.HashSet;
  51 import java.util.Iterator;
  52 import java.util.Set;
  53
  54 public class SelectorImpl extends AbstractSelector
  55 {
  56   private Set keys;
  57   private Set selected;
  58
  59   /**
  60    * A dummy object whose monitor regulates access to both our
  61    * selectThread and unhandledWakeup fields.
  62    */
  63   private Object selectThreadMutex = new Object ();
  64
  65   /**
  66    * Any thread that's currently blocked in a select operation.
  67    */
  68   private Thread selectThread;
  69
  70   /**
  71    * Indicates whether we have an unhandled wakeup call. This can
  72    * be due to either wakeup() triggering a thread interruption while
  73    * a thread was blocked in a select operation (in which case we need
  74    * to reset this thread's interrupt status after interrupting the
  75    * select), or else that no thread was on a select operation at the
  76    * time that wakeup() was called, in which case the following select()
  77    * operation should return immediately with nothing selected.
  78    */
  79   private boolean unhandledWakeup;
  80
  81   public SelectorImpl (SelectorProvider provider)
  82   {
  83     super (provider);
  84
  85     keys = new HashSet ();
  86     selected = new HashSet ();
  87   }
  88
  89   protected void finalize() throws Throwable
  90   {
  91     close();
  92   }
  93
  94   protected final void implCloseSelector()
  95     throws IOException
  96   {
  97     // Cancel any pending select operation.
  98     wakeup();
  99
 100     synchronized (keys)
 101       {
 102         synchronized (selected)
 103           {
 104             synchronized (cancelledKeys ())
 105               {
 106                 // FIXME: Release resources here.
 107               }
 108           }
 109       }
 110   }
 111
 112   public final Set keys()
 113   {
 114     if (!isOpen())
 115       throw new ClosedSelectorException();
 116
 117     return Collections.unmodifiableSet (keys);
 118   }
 119
 120   public final int selectNow()
 121     throws IOException
 122   {
 123     // FIXME: We're simulating an immediate select
 124     // via a select with a timeout of one millisecond.
 125     return select (1);
 126   }
 127
 128   public final int select()
 129     throws IOException
 130   {
 131     return select (0);
 132   }
 133
 134   private final int[] getFDsAsArray (int ops)
 135   {
 136     int[] result;
 137     int counter = 0;
 138     Iterator it = keys.iterator ();
 139
 140     // Count the number of file descriptors needed
 141     while (it.hasNext ())
 142       {
 143         SelectionKeyImpl key = (SelectionKeyImpl) it.next ();
 144
 145         if ((key.interestOps () & ops) != 0)
 146           {
 147             counter++;
 148           }
 149       }
 150
 151     result = new int[counter];
 152
 153     counter = 0;
 154     it = keys.iterator ();
 155
 156     // Fill the array with the file descriptors
 157     while (it.hasNext ())
 158       {
 159         SelectionKeyImpl key = (SelectionKeyImpl) it.next ();
 160
 161         if ((key.interestOps () & ops) != 0)
 162           {
 163             result[counter] = key.getNativeFD();
 164             counter++;
 165           }
 166       }
 167
 168     return result;
 169   }
 170
 171   public synchronized int select (long timeout)
 172     throws IOException
 173   {
 174     if (!isOpen())
 175       throw new ClosedSelectorException();
 176
 177     synchronized (keys)
 178       {
 179         synchronized (selected)
 180           {
 181             deregisterCancelledKeys();
 182
 183             // Set only keys with the needed interest ops into the arrays.
 184             int[] read = getFDsAsArray (SelectionKey.OP_READ
 185                                         | SelectionKey.OP_ACCEPT);
 186             int[] write = getFDsAsArray (SelectionKey.OP_WRITE
 187                                          | SelectionKey.OP_CONNECT);
 188
 189             // FIXME: We dont need to check this yet
 190             int[] except = new int [0];
 191
 192             // Test to see if we've got an unhandled wakeup call,
 193             // in which case we return immediately. Otherwise,
 194             // remember our current thread and jump into the select.
 195             // The monitor for dummy object selectThreadMutex regulates
 196             // access to these fields.
 197
 198             // FIXME: Not sure from the spec at what point we should
 199             // return "immediately". Is it here or immediately upon
 200             // entry to this function?
 201
 202             // NOTE: There's a possibility of another thread calling
 203             // wakeup() immediately after our thread releases
 204             // selectThreadMutex's monitor here, in which case we'll
 205             // do the select anyway. Since calls to wakeup() and select()
 206             // among different threads happen in non-deterministic order,
 207             // I don't think this is an issue.
 208             synchronized (selectThreadMutex)
 209               {
 210                 if (unhandledWakeup)
 211                   {
 212                     unhandledWakeup = false;
 213                     return 0;
 214                   }
 215                 else
 216                   {
 217                     selectThread = Thread.currentThread ();
 218                   }
 219               }
 220
 221             // Call the native select() on all file descriptors.
 222             int result = 0;
 223             try
 224               {
 225                 begin();
 226                 result = VMSelector.select (read, write, except, timeout);
 227               }
 228             finally
 229               {
 230                 end();
 231               }
 232
 233             // If our unhandled wakeup flag is set at this point,
 234             // reset our thread's interrupt flag because we were
 235             // awakened by wakeup() instead of an external thread
 236             // interruption.
 237             //
 238             // NOTE: If we were blocked in a select() and one thread
 239             // called Thread.interrupt() on the blocked thread followed
 240             // by another thread calling Selector.wakeup(), then race
 241             // conditions could make it so that the thread's interrupt
 242             // flag is reset even though the Thread.interrupt() call
 243             // "was there first". I don't think we need to care about
 244             // this scenario.
 245             synchronized (selectThreadMutex)
 246               {
 247                 if (unhandledWakeup)
 248                   {
 249                     unhandledWakeup = false;
 250                     Thread.interrupted ();
 251                   }
 252                 selectThread = null;
 253               }
 254
 255             Iterator it = keys.iterator ();
 256
 257             while (it.hasNext ())
 258               {
 259                 int ops = 0;
 260                 SelectionKeyImpl key = (SelectionKeyImpl) it.next ();
 261
 262                 // If key is already selected retrieve old ready ops.
 263                 if (selected.contains (key))
 264                   {
 265                     ops = key.readyOps ();
 266                   }
 267
 268                 // Set new ready read/accept ops
 269                 for (int i = 0; i < read.length; i++)
 270                   {
 271                     if (key.getNativeFD() == read[i])
 272                       {
 273                         if (key.channel () instanceof ServerSocketChannelImpl)
 274                           {
 275                             ops = ops | SelectionKey.OP_ACCEPT;
 276                           }
 277                         else
 278                           {
 279                             ops = ops | SelectionKey.OP_READ;
 280                           }
 281                       }
 282                   }
 283
 284                 // Set new ready write ops
 285                 for (int i = 0; i < write.length; i++)
 286                   {
 287                     if (key.getNativeFD() == write[i])
 288                       {
 289                         ops = ops | SelectionKey.OP_WRITE;
 290
 291         //                 if (key.channel ().isConnected ())
 292         //                   {
 293         //                     ops = ops | SelectionKey.OP_WRITE;
 294         //                   }
 295         //                 else
 296         //                   {
 297         //                     ops = ops | SelectionKey.OP_CONNECT;
 298         //                   }
 299                      }
 300                   }
 301
 302                 // FIXME: We dont handle exceptional file descriptors yet.
 303
 304                 // If key is not yet selected add it.
 305                 if (!selected.contains (key))
 306                   {
 307                     selected.add (key);
 308                   }
 309
 310                 // Set new ready ops
 311                 key.readyOps (key.interestOps () & ops);
 312               }
 313             deregisterCancelledKeys();
 314
 315             return result;
 316           }
 317         }
 318   }
 319
 320   public final Set selectedKeys()
 321   {
 322     if (!isOpen())
 323       throw new ClosedSelectorException();
 324
 325     return selected;
 326   }
 327
 328   public final Selector wakeup()
 329   {
 330     // IMPLEMENTATION NOTE: Whereas the specification says that
 331     // thread interruption should trigger a call to wakeup, we
 332     // do the reverse under the covers: wakeup triggers a thread
 333     // interrupt followed by a subsequent reset of the thread's
 334     // interrupt status within select().
 335
 336     // First, acquire the monitor of the object regulating
 337     // access to our selectThread and unhandledWakeup fields.
 338     synchronized (selectThreadMutex)
 339       {
 340         unhandledWakeup = true;
 341
 342         // Interrupt any thread which is currently blocked in
 343         // a select operation.
 344         if (selectThread != null)
 345           selectThread.interrupt ();
 346       }
 347
 348     return this;
 349   }
 350
 351   private final void deregisterCancelledKeys()
 352   {
 353     Set ckeys = cancelledKeys ();
 354     synchronized (ckeys)
 355     {
 356       Iterator it = ckeys.iterator();
 357
 358       while (it.hasNext ())
 359         {
 360           keys.remove ((SelectionKeyImpl) it.next ());
 361           it.remove ();
 362         }
 363     }
 364   }
 365
 366   protected SelectionKey register (SelectableChannel ch, int ops, Object att)
 367   {
 368     return register ((AbstractSelectableChannel) ch, ops, att);
 369   }
 370
 371   protected final SelectionKey register (AbstractSelectableChannel ch, int ops,
 372                                          Object att)
 373   {
 374     SelectionKeyImpl result;
 375
 376     if (ch instanceof SocketChannelImpl)
 377       result = new SocketChannelSelectionKey (ch, this);
 378     else if (ch instanceof DatagramChannelImpl)
 379       result = new DatagramChannelSelectionKey (ch, this);
 380     else if (ch instanceof ServerSocketChannelImpl)
 381       result = new ServerSocketChannelSelectionKey (ch, this);
 382     else
 383       throw new InternalError ("No known channel type");
 384
 385     synchronized (keys)
 386       {
 387         keys.add (result);
 388       }
 389
 390     result.interestOps (ops);
 391     result.attach (att);
 392     return result;
 393   }
 394 }