BlockingIO.java

  1. /*
  2.  **** BEGIN LICENSE BLOCK *****
  3.  * Version: EPL 1.0/GPL 2.0/LGPL 2.1
  4.  *
  5.  * The contents of this file are subject to the Eclipse Public
  6.  * License Version 1.0 (the "License"); you may not use this file
  7.  * except in compliance with the License. You may obtain a copy of
  8.  * the License at http://www.eclipse.org/legal/epl-v10.html
  9.  *
  10.  * Software distributed under the License is distributed on an "AS
  11.  * IS" basis, WITHOUT WARRANTY OF ANY KIND, either express or
  12.  * implied. See the License for the specific language governing
  13.  * rights and limitations under the License.
  14.  *
  15.  * Copyright (C) 2008 The JRuby Community <www.jruby.org>
  16.  * 
  17.  * Alternatively, the contents of this file may be used under the terms of
  18.  * either of the GNU General Public License Version 2 or later (the "GPL"),
  19.  * or the GNU Lesser General Public License Version 2.1 or later (the "LGPL"),
  20.  * in which case the provisions of the GPL or the LGPL are applicable instead
  21.  * of those above. If you wish to allow use of your version of this file only
  22.  * under the terms of either the GPL or the LGPL, and not to allow others to
  23.  * use your version of this file under the terms of the EPL, indicate your
  24.  * decision by deleting the provisions above and replace them with the notice
  25.  * and other provisions required by the GPL or the LGPL. If you do not delete
  26.  * the provisions above, a recipient may use your version of this file under
  27.  * the terms of any one of the EPL, the GPL or the LGPL.
  28.  ***** END LICENSE BLOCK *****/

  30. package org.jruby.util.io;

  32. import java.io.IOException;
  33. import java.io.InterruptedIOException;
  34. import java.nio.ByteBuffer;
  35. import java.nio.channels.Channel;
  36. import java.nio.channels.ReadableByteChannel;
  37. import java.nio.channels.SelectableChannel;
  38. import java.nio.channels.SelectionKey;
  39. import java.nio.channels.Selector;
  40. import java.nio.channels.WritableByteChannel;
  41. import java.nio.channels.spi.SelectorProvider;
  42. import java.util.HashSet;
  43. import java.util.LinkedList;
  44. import java.util.List;
  45. import java.util.Map;
  46. import java.util.Set;
  47. import java.util.concurrent.ConcurrentHashMap;
  48. import java.util.concurrent.ConcurrentLinkedQueue;
  49. import java.util.concurrent.TimeUnit;

  51. /**
  52.  * A Utility class to emulate blocking I/O operations on non-blocking channels.
  53.  */
  54. public class BlockingIO {
  55.     public static final class Condition {
  56.         private final IOChannel channel;
  57.         Condition(IOChannel channel) {
  58.             this.channel = channel;
  59.         }
  60.         public void cancel() {
  61.             channel.wakeup(false);
  62.         }
  63.         public void interrupt() {
  64.             channel.interrupt();
  65.         }
  66.         public boolean await() throws InterruptedException {
  67.             return channel.await();
  68.         }
  69.         public boolean await(long timeout, TimeUnit unit) throws InterruptedException {
  70.             return channel.await(timeout, unit);
  71.         }
  72.     }
  73.     static final class IOChannel {
  74.         final SelectableChannel channel;
  75.         final int ops;        
  76.         private final Object monitor;
  77.         private boolean woken = false;
  78.         private boolean ready = false;
  79.         private boolean interrupted = false;
  80.         
  81.         IOChannel(SelectableChannel channel, int ops, Object monitor) {
  82.             this.channel = channel;
  83.             this.ops = ops;
  84.             this.monitor = monitor;
  85.         }
  86.         public final void wakeup(boolean ready) {
  87.             synchronized (monitor) {
  88.                 this.woken = true;
  89.                 this.ready = ready;
  90.                 monitor.notifyAll();
  91.             }
  92.         }
  93.         public final void interrupt() {
  94.             synchronized (monitor) {
  95.                 this.woken = true;
  96.                 this.interrupted = true;
  97.                 monitor.notifyAll();
  98.             }
  99.         }
  100.         public final boolean await() throws InterruptedException {
  101.             return await(0, TimeUnit.MILLISECONDS);
  102.         }
  103.         public final boolean await(final long timeout, TimeUnit unit) throws InterruptedException {
  104.             synchronized (monitor) {
  105.                 if (!woken) {
  106.                     monitor.wait(TimeUnit.MILLISECONDS.convert(timeout, unit));
  107.                 }
  108.                 if (interrupted) {
  109.                     throw new InterruptedException("Interrupted");
  110.                 }
  111.                 return ready;
  112.             }
  113.         }
  114.     }
  115.     static final class IOSelector implements Runnable {
  116.         private final Selector selector;
  117.         private final ConcurrentLinkedQueue<IOChannel> registrationQueue;

  119.         public IOSelector(SelectorProvider provider) throws IOException {
  120.             selector = SelectorFactory.openWithRetryFrom(null, provider);
  121.             registrationQueue = new ConcurrentLinkedQueue<IOChannel>();
  122.         }
  123.         public void run() {
  124.             for ( ; ; ) {
  125.                 try {
  126.                     //
  127.                     // Wake up any channels that became unblocked
  128.                     //
  129.                     Set<SelectionKey> selected = new HashSet<SelectionKey>(selector.selectedKeys());
  130.                     for (SelectionKey k : selected) {
  131.                         List<IOChannel> waitq = (List<IOChannel>) k.attachment();
  132.                         for (IOChannel ch : waitq) {
  133.                             ch.wakeup(true);
  134.                         }
  135.                         waitq.clear();
  136.                     }

  138.                     //
  139.                     // Register any new blocking I/O requests
  140.                     //
  141.                     IOChannel ch;
  142.                     Set<SelectableChannel> added = new HashSet<SelectableChannel>();
  143.                     while ((ch = registrationQueue.poll()) != null) {
  144.                         SelectionKey k = ch.channel.keyFor(selector);
  145.                         List<IOChannel> waitq = k == null
  146.                                 ? new LinkedList<IOChannel>()
  147.                                 : (List<IOChannel>) k.attachment();
  148.                         ch.channel.register(selector, ch.ops, waitq);
  149.                         waitq.add(ch);
  150.                         added.add(ch.channel);
  151.                     }

  153.                     // Now clear out any previously selected channels
  154.                     for (SelectionKey k : selected) {
  155.                         if (!added.contains(k.channel())) {
  156.                             k.cancel();
  157.                         }
  158.                     }

  160.                     //
  161.                     // Wait for I/O on any channel
  162.                     //
  163.                     selector.select();
  164.                 } catch (IOException ex) {

  166.                 }
  167.             }
  168.         }
  169.         Condition add(Channel channel, int ops, Object monitor) {
  170.             IOChannel io = new IOChannel((SelectableChannel) channel, ops, monitor);
  171.             registrationQueue.add(io);
  172.             selector.wakeup();
  173.             return new Condition(io);
  174.         }
  175.         public void await(Channel channel, int op) throws InterruptedException {
  176.             add(channel, op, new Object()).await();
  177.         }
  178.     }
  179.     static final private Map<SelectorProvider, IOSelector> selectors
  180.             = new ConcurrentHashMap<SelectorProvider, IOSelector>();

  182.     private static IOSelector getSelector(SelectorProvider provider) throws IOException {
  183.         IOSelector sel = selectors.get(provider);
  184.         if (sel != null) {
  185.             return sel;
  186.         }

  188.         //
  189.         // Synchronize and re-check to avoid creating more than one Selector per provider
  190.         //
  191.         synchronized (provider) {
  192.             sel = selectors.get(provider);
  193.             if (sel == null) {
  194.                 sel = new IOSelector(provider);
  195.                 selectors.put(provider, sel);
  196.                 Thread t = new Thread(sel);
  197.                 t.setDaemon(true);
  198.                 t.start();
  199.             }
  200.         }
  201.         return sel;
  202.     }
  203.     private static IOSelector getSelector(Channel channel) throws IOException {
  204.         if (!(channel instanceof SelectableChannel)) {
  205.             throw new IllegalArgumentException("channel must be a SelectableChannel");
  206.         }        
  207.         return getSelector(((SelectableChannel) channel).provider());
  208.     }
  209.     public static final Condition newCondition(Channel channel, int ops, Object monitor) throws IOException {
  210.         return getSelector(channel).add(channel, ops, monitor);
  211.     }
  212.     public static final Condition newCondition(Channel channel, int ops) throws IOException {
  213.         return newCondition(channel, ops, new Object());
  214.     }
  215.     public static void waitForIO(Channel channel, int op) throws InterruptedException, IOException {
  216.         getSelector(channel).await(channel, op);
  217.     }
  218.     public static void awaitReadable(ReadableByteChannel channel) throws InterruptedException, IOException {
  219.         waitForIO(channel, SelectionKey.OP_READ);
  220.     }
  221.     public static void awaitWritable(WritableByteChannel channel) throws InterruptedException, IOException {
  222.         waitForIO(channel, SelectionKey.OP_WRITE);
  223.     }
  224.     public static int read(ReadableByteChannel channel, ByteBuffer buf, boolean blocking) throws IOException {
  225.         do {
  226.             int n = channel.read(buf);
  227.             if (n != 0 || !blocking || !(channel instanceof SelectableChannel) || !buf.hasRemaining()) {
  228.                 return n;
  229.             }
  230.             try {
  231.                 awaitReadable(channel);
  232.             } catch (InterruptedException ex) {
  233.                 throw new InterruptedIOException(ex.getMessage());
  234.             }
  235.         } while (true);
  236.     }
  237.     public static int write(WritableByteChannel channel, ByteBuffer buf, boolean blocking) throws IOException {
  238.         do {
  239.             int n = channel.write(buf);
  240.             if (n != 0 || !blocking || !(channel instanceof SelectableChannel) || !buf.hasRemaining()) {
  241.                 return n;
  242.             }
  243.             try {
  244.                 awaitWritable(channel);
  245.             } catch (InterruptedException ex) {
  246.                 throw new InterruptedIOException(ex.getMessage());
  247.             }
  248.         } while (true);
  249.     }
  250.     public static int blockingRead(ReadableByteChannel channel, ByteBuffer buf) throws IOException {
  251.         return read(channel, buf, true);
  252.     }
  253.     public static int blockingWrite(WritableByteChannel channel, ByteBuffer buf) throws IOException {
  254.         return write(channel, buf, true);
  255.     }
  256. }
Created with JaCoCo 0.7.3.201501050207