NioSelectorPool


1   /*
2    *  Copyright 2005-2006 The Apache Software Foundation
3    *
4    *  Licensed under the Apache License, Version 2.0 (the "License");
5    *  you may not use this file except in compliance with the License.
6    *  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,
12   *  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13   *  See the License for the specific language governing permissions and
14   *  limitations under the License.
15   */
16  package org.apache.tomcat.util.net;
17  
18  import java.util.LinkedList  ;
19  import java.util.concurrent.atomic.AtomicInteger  ;
20  import java.nio.channels.Selector  ;
21  import java.io.IOException  ;
22  import java.util.NoSuchElementException  ;
23  import java.nio.ByteBuffer  ;
24  import java.nio.channels.SocketChannel  ;
25  import java.nio.channels.SelectionKey  ;
26  import java.io.EOFException  ;
27  import java.net.SocketTimeoutException  ;
28  import java.util.concurrent.ConcurrentLinkedQueue  ;
29  
30  /**
31   * 
32   * Thread safe non blocking selector pool
33   * @author Filip Hanik
34   * @version 1.0
35   * @since 6.0
36   */
37  
38  public class NioSelectorPool {
39      protected int maxSelectors = 200;
40      protected int maxSpareSelectors = -1;
41      protected boolean enabled = true;
42      protected AtomicInteger   active = new AtomicInteger  (0);
43      protected AtomicInteger   spare = new AtomicInteger  (0);
44      //protected LinkedList<Selector> selectors = new LinkedList<Selector>();
45      protected ConcurrentLinkedQueue  <Selector  > selectors = new ConcurrentLinkedQueue  <Selector  >();
46      
47      public Selector   get() throws IOException  {
48          if ( (!enabled) || active.incrementAndGet() >= maxSelectors ) {
49              if ( enabled ) active.decrementAndGet();
50              return null;
51          }
52          Selector   s = null;
53          try {
54              s = selectors.size()>0?selectors.poll():null;
55              if (s == null) s = Selector.open();
56              else spare.decrementAndGet();
57              
58          }catch (NoSuchElementException   x ) {
59              try {s = Selector.open();}catch (IOException   iox){}
60          } finally {
61              if ( s == null ) active.decrementAndGet();//we were unable to find a selector
62          }
63          return s;            
64      }
65      
66      
67      
68      public void put(Selector   s) throws IOException   {
69          if ( enabled ) active.decrementAndGet();
70          if ( enabled && (maxSpareSelectors==-1 || spare.get() < Math.min(maxSpareSelectors,maxSelectors)) ) {
71              spare.incrementAndGet();
72              selectors.offer(s);
73          }
74          else s.close();
75      }
76      
77      public void close() throws IOException   {
78          enabled = false;
79          Selector   s;
80          while ( (s = selectors.poll()) != null ) s.close();
81          spare.set(0);
82          active.set(0);
83      }
84      
85      public void open(){
86          enabled = true;
87      }
88      
89      /**
90       * Performs a blocking write using the bytebuffer for data to be written and a selector to block.
91       * If the <code>selector</code> parameter is null, then it will perform a busy write that could
92       * take up a lot of CPU cycles.
93       * @param buf ByteBuffer - the buffer containing the data, we will write as long as <code>(buf.hasRemaining()==true)</code>
94       * @param socket SocketChannel - the socket to write data to
95       * @param selector Selector - the selector to use for blocking, if null then a busy write will be initiated
96       * @param writeTimeout long - the timeout for this write operation in milliseconds, -1 means no timeout
97       * @return int - returns the number of bytes written
98       * @throws EOFException if write returns -1
99       * @throws SocketTimeoutException if the write times out
100      * @throws IOException if an IO Exception occurs in the underlying socket logic
101      */
102     public int write(ByteBuffer   buf, NioChannel socket, Selector   selector, long writeTimeout) throws IOException   {
103         SelectionKey   key = null;
104         int written = 0;
105         boolean timedout = false;
106         int keycount = 1; //assume we can write
107         long time = System.currentTimeMillis(); //start the timeout timer
108         try {
109             while ( (!timedout) && buf.hasRemaining() ) {
110                 if ( keycount > 0 ) { //only write if we were registered for a write
111                     int cnt = socket.write(buf); //write the data
112                     if (cnt == -1) throw new EOFException  ();
113                     written += cnt;
114                     if (cnt > 0) {
115                         time = System.currentTimeMillis(); //reset our timeout timer
116                         continue; //we successfully wrote, try again without a selector
117                     }
118                 }
119                 if ( selector != null ) {
120                     //register OP_WRITE to the selector
121                     if (key==null) key = socket.getIOChannel().register(selector, SelectionKey.OP_WRITE);
122                     else key.interestOps(SelectionKey.OP_WRITE);
123                     keycount = selector.select(writeTimeout);
124                 }                
125                 if (writeTimeout > 0 && (selector == null || keycount == 0) ) timedout = (System.currentTimeMillis()-time)>=writeTimeout;
126             }//while
127             if ( timedout ) throw new SocketTimeoutException  ();
128         } finally {
129             if (key != null) {
130                 key.cancel();
131                 if (selector != null) selector.selectNow();//removes the key from this selector
132             }
133         }
134         return written;
135     }
136      
137     /**
138      * Performs a blocking read using the bytebuffer for data to be read and a selector to block.
139      * If the <code>selector</code> parameter is null, then it will perform a busy read that could
140      * take up a lot of CPU cycles.
141      * @param buf ByteBuffer - the buffer containing the data, we will read as until we have read at least one byte or we timed out
142      * @param socket SocketChannel - the socket to write data to
143      * @param selector Selector - the selector to use for blocking, if null then a busy read will be initiated
144      * @param readTimeout long - the timeout for this read operation in milliseconds, -1 means no timeout
145      * @return int - returns the number of bytes read
146      * @throws EOFException if read returns -1
147      * @throws SocketTimeoutException if the read times out
148      * @throws IOException if an IO Exception occurs in the underlying socket logic
149      */
150     public int read(ByteBuffer   buf, NioChannel socket, Selector   selector, long readTimeout) throws IOException   {
151         SelectionKey   key = null;
152         int read = 0;
153         boolean timedout = false;
154         int keycount = 1; //assume we can write
155         long time = System.currentTimeMillis(); //start the timeout timer
156         try {
157             while ( (!timedout) && read == 0 ) {
158                 if ( keycount > 0 ) { //only read if we were registered for a read
159                     int cnt = socket.read(buf);
160                     if (cnt == -1) throw new EOFException  ();
161                     read += cnt;
162                     if (cnt > 0) break;
163                 }
164                 if ( selector != null ) {
165                     //register OP_WRITE to the selector
166                     if (key==null) key = socket.getIOChannel().register(selector, SelectionKey.OP_READ);
167                     else key.interestOps(SelectionKey.OP_READ);
168                     keycount = selector.select(readTimeout);
169                 }                
170                 if (readTimeout > 0 && (selector == null || keycount == 0) ) timedout = (System.currentTimeMillis()-time)>=readTimeout;
171             }//while
172             if ( timedout ) throw new SocketTimeoutException  ();
173         } finally {
174             if (key != null) {
175                 key.cancel();
176                 if (selector != null) selector.selectNow();//removes the key from this selector
177             }
178         }
179         return read;
180     }
181     
182     public void setMaxSelectors(int maxSelectors) {
183         this.maxSelectors = maxSelectors;
184     }
185 
186     public void setMaxSpareSelectors(int maxSpareSelectors) {
187         this.maxSpareSelectors = maxSpareSelectors;
188     }
189 
190     public void setEnabled(boolean enabled) {
191         this.enabled = enabled;
192     }
193 
194     public int getMaxSelectors() {
195         return maxSelectors;
196     }
197 
198     public int getMaxSpareSelectors() {
199         return maxSpareSelectors;
200     }
201 
202     public boolean isEnabled() {
203         return enabled;
204     }
205 }
A to Z: JavaDoc & Examples Daily Java News & Articles Open Source Projects Open Source Codes Free Computer Books Remove Frame
Popular Tags