Java > Open Source Codes > org > apache > tomcat > util > net > PoolTcpEndpoint


1 /*
2  * Licensed to the Apache Software Foundation (ASF) under one or more
3  * contributor license agreements. See the NOTICE file distributed with
4  * this work for additional information regarding copyright ownership.
5  * The ASF licenses this file to You under the Apache License, Version 2.0
6  * (the "License"); you may not use this file except in compliance with
7  * the License. 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.apache.tomcat.util.net;
19
20 import java.io.IOException ;
21 import java.io.InterruptedIOException ;
22 import java.net.BindException ;
23 import java.net.InetAddress ;
24 import java.net.ServerSocket ;
25 import java.net.Socket ;
26 import java.net.SocketException ;
27 import java.security.AccessControlException ;
28 import java.util.Stack ;
29 import java.util.Vector ;
30
31 import org.apache.commons.logging.Log;
32 import org.apache.commons.logging.LogFactory;
33 import org.apache.tomcat.util.res.StringManager;
34 import org.apache.tomcat.util.threads.ThreadPool;
35 import org.apache.tomcat.util.threads.ThreadPoolRunnable;
36
37 /* Similar with MPM module in Apache2.0. Handles all the details related with
38    "tcp server" functionality - thread management, accept policy, etc.
39    It should do nothing more - as soon as it get a socket ( and all socket options
40    are set, etc), it just handle the stream to ConnectionHandler.processConnection. (costin)
41 */
42
43
44
45 /**
46  * Handle incoming TCP connections.
47  *
48  * This class implement a simple server model: one listener thread accepts on a socket and
49  * creates a new worker thread for each incoming connection.
50  *
51  * More advanced Endpoints will reuse the threads, use queues, etc.
52  *
53  * @author James Duncan Davidson [duncan@eng.sun.com]
54  * @author Jason Hunter [jch@eng.sun.com]
55  * @author James Todd [gonzo@eng.sun.com]
56  * @author Costin@eng.sun.com
57  * @author Gal Shachor [shachor@il.ibm.com]
58  * @author Yoav Shapira <yoavs@apache.org>
59  */
60 public class PoolTcpEndpoint implements Runnable { // implements Endpoint {
61
62     static Log log=LogFactory.getLog(PoolTcpEndpoint.class );
63
64     private StringManager sm =
65         StringManager.getManager("org.apache.tomcat.util.net.res");
66
67     private static final int BACKLOG = 100;
68     private static final int TIMEOUT = 1000;
69
70     private final Object threadSync = new Object ();
71
72     private int backlog = BACKLOG;
73     private int serverTimeout = TIMEOUT;
74
75     private InetAddress inet;
76     private int port;
77
78     private ServerSocketFactory factory;
79     private ServerSocket serverSocket;
80
81     private volatile boolean running = false;
82     private volatile boolean paused = false;
83     private boolean initialized = false;
84     private boolean reinitializing = false;
85     static final int debug=0;
86
87     protected boolean tcpNoDelay=false;
88     protected int linger=100;
89     protected int socketTimeout=-1;
90     private boolean lf = true;
91
92     
93     // ------ Leader follower fields
94
95     
96     TcpConnectionHandler handler;
97     ThreadPoolRunnable listener;
98     ThreadPool tp;
99
100     
101     // ------ Master slave fields
102
103     /* The background thread. */
104     private Thread thread = null;
105     /* Available processors. */
106     private Stack workerThreads = new Stack ();
107     private int curThreads = 0;
108     private int maxThreads = 20;
109     /* All processors which have been created. */
110     private Vector created = new Vector ();
111
112     
113     public PoolTcpEndpoint() {
114     tp = new ThreadPool();
115     }
116
117     public PoolTcpEndpoint( ThreadPool tp ) {
118         this.tp=tp;
119     }
120
121     // -------------------- Configuration --------------------
122
123     public void setMaxThreads(int maxThreads) {
124     if( maxThreads > 0)
125         tp.setMaxThreads(maxThreads);
126     }
127
128     public int getMaxThreads() {
129         return tp.getMaxThreads();
130     }
131
132     public void setMaxSpareThreads(int maxThreads) {
133     if(maxThreads > 0)
134         tp.setMaxSpareThreads(maxThreads);
135     }
136
137     public int getMaxSpareThreads() {
138         return tp.getMaxSpareThreads();
139     }
140
141     public void setMinSpareThreads(int minThreads) {
142     if(minThreads > 0)
143         tp.setMinSpareThreads(minThreads);
144     }
145
146     public int getMinSpareThreads() {
147         return tp.getMinSpareThreads();
148     }
149
150     public void setThreadPriority(int threadPriority) {
151       tp.setThreadPriority(threadPriority);
152     }
153
154     public int getThreadPriority() {
155       return tp.getThreadPriority();
156     }
157
158     public int getPort() {
159         return port;
160     }
161
162     public void setPort(int port ) {
163         this.port=port;
164     }
165
166     public InetAddress getAddress() {
167         return inet;
168     }
169
170     public void setAddress(InetAddress inet) {
171         this.inet=inet;
172     }
173
174     public void setServerSocket(ServerSocket ss) {
175         serverSocket = ss;
176     }
177
178     public void setServerSocketFactory( ServerSocketFactory factory ) {
179         this.factory=factory;
180     }
181
182    ServerSocketFactory getServerSocketFactory() {
183         return factory;
184    }
185
186     public void setConnectionHandler( TcpConnectionHandler handler ) {
187         this.handler=handler;
188     }
189
190     public TcpConnectionHandler getConnectionHandler() {
191         return handler;
192     }
193
194     public boolean isRunning() {
195     return running;
196     }
197     
198     public boolean isPaused() {
199     return paused;
200     }
201     
202     /**
203      * Allows the server developer to specify the backlog that
204      * should be used for server sockets. By default, this value
205      * is 100.
206      */
207     public void setBacklog(int backlog) {
208     if( backlog>0)
209         this.backlog = backlog;
210     }
211
212     public int getBacklog() {
213         return backlog;
214     }
215
216     /**
217      * Sets the timeout in ms of the server sockets created by this
218      * server. This method allows the developer to make servers
219      * more or less responsive to having their server sockets
220      * shut down.
221      *
222      * <p>By default this value is 1000ms.
223      */
224     public void setServerTimeout(int timeout) {
225     this.serverTimeout = timeout;
226     }
227
228     public boolean getTcpNoDelay() {
229         return tcpNoDelay;
230     }
231     
232     public void setTcpNoDelay( boolean b ) {
233     tcpNoDelay=b;
234     }
235
236     public int getSoLinger() {
237         return linger;
238     }
239     
240     public void setSoLinger( int i ) {
241     linger=i;
242     }
243
244     public int getSoTimeout() {
245         return socketTimeout;
246     }
247     
248     public void setSoTimeout( int i ) {
249     socketTimeout=i;
250     }
251     
252     public int getServerSoTimeout() {
253         return serverTimeout;
254     }
255     
256     public void setServerSoTimeout( int i ) {
257     serverTimeout=i;
258     }
259
260     public String getStrategy() {
261         if (lf) {
262             return "lf";
263         } else {
264             return "ms";
265         }
266     }
267     
268     public void setStrategy(String strategy) {
269         if ("ms".equals(strategy)) {
270             lf = false;
271         } else {
272             lf = true;
273         }
274     }
275
276     public int getCurrentThreadCount() {
277         return curThreads;
278     }
279     
280     public int getCurrentThreadsBusy() {
281         return curThreads - workerThreads.size();
282     }
283     
284     // -------------------- Public methods --------------------
285
286     public void initEndpoint() throws IOException , InstantiationException {
287         try {
288             if(factory==null)
289                 factory=ServerSocketFactory.getDefault();
290             if(serverSocket==null) {
291                 try {
292                     if (inet == null) {
293                         serverSocket = factory.createSocket(port, backlog);
294                     } else {
295                         serverSocket = factory.createSocket(port, backlog, inet);
296                     }
297                 } catch ( BindException be ) {
298                     throw new BindException (be.getMessage() + ":" + port);
299                 }
300             }
301             if( serverTimeout >= 0 )
302                 serverSocket.setSoTimeout( serverTimeout );
303         } catch( IOException ex ) {
304             throw ex;
305         } catch( InstantiationException ex1 ) {
306             throw ex1;
307         }
308         initialized = true;
309     }
310     
311     public void startEndpoint() throws IOException , InstantiationException {
312         if (!initialized) {
313             initEndpoint();
314         }
315         if (lf) {
316             tp.start();
317         }
318         running = true;
319         paused = false;
320         if (lf) {
321             listener = new LeaderFollowerWorkerThread(this);
322             tp.runIt(listener);
323         } else {
324             maxThreads = getMaxThreads();
325             threadStart();
326         }
327     }
328
329     public void pauseEndpoint() {
330         if (running && !paused) {
331             paused = true;
332             unlockAccept();
333         }
334     }
335
336     public void resumeEndpoint() {
337         if (running) {
338             paused = false;
339         }
340     }
341
342     public void stopEndpoint() {
343         if (running) {
344             if (lf) {
345                 tp.shutdown();
346             }
347             running = false;
348             if (serverSocket != null) {
349                 closeServerSocket();
350             }
351             if (!lf) {
352                 threadStop();
353             }
354             initialized=false ;
355         }
356     }
357
358     protected void closeServerSocket() {
359         if (!paused)
360             unlockAccept();
361         try {
362             if( serverSocket!=null)
363                 serverSocket.close();
364         } catch(Exception e) {
365             log.error(sm.getString("endpoint.err.close"), e);
366         }
367         serverSocket = null;
368     }
369
370     protected void unlockAccept() {
371         Socket s = null;
372         try {
373             // Need to create a connection to unlock the accept();
374 if (inet == null) {
375                 s = new Socket ("127.0.0.1", port);
376             } else {
377                 s = new Socket (inet, port);
378                     // setting soLinger to a small value will help shutdown the
379 // connection quicker
380 s.setSoLinger(true, 0);
381             }
382         } catch(Exception e) {
383             if (log.isDebugEnabled()) {
384                 log.debug(sm.getString("endpoint.debug.unlock", "" + port), e);
385             }
386         } finally {
387             if (s != null) {
388                 try {
389                     s.close();
390                 } catch (Exception e) {
391                     // Ignore
392 }
393             }
394         }
395     }
396
397     // -------------------- Private methods
398
399     Socket acceptSocket() {
400         if( !running || serverSocket==null ) return null;
401
402         Socket accepted = null;
403
404         try {
405             if(factory==null) {
406                 accepted = serverSocket.accept();
407             } else {
408                 accepted = factory.acceptSocket(serverSocket);
409             }
410             if (null == accepted) {
411                 log.warn(sm.getString("endpoint.warn.nullSocket"));
412             } else {
413                 if (!running) {
414                     accepted.close(); // rude, but unlikely!
415 accepted = null;
416                 } else if (factory != null) {
417                     factory.initSocket( accepted );
418                 }
419             }
420         }
421         catch(InterruptedIOException iioe) {
422             // normal part -- should happen regularly so
423 // that the endpoint can release if the server
424 // is shutdown.
425 }
426         catch (AccessControlException ace) {
427             // When using the Java SecurityManager this exception
428 // can be thrown if you are restricting access to the
429 // socket with SocketPermission's.
430 // Log the unauthorized access and continue
431 String msg = sm.getString("endpoint.warn.security",
432                                       serverSocket, ace);
433             log.warn(msg);
434         }
435         catch (IOException e) {
436
437             String msg = null;
438
439             if (running) {
440                 msg = sm.getString("endpoint.err.nonfatal",
441                         serverSocket, e);
442                 log.error(msg, e);
443             }
444
445             if (accepted != null) {
446                 try {
447                     accepted.close();
448                 } catch(Throwable ex) {
449                     msg = sm.getString("endpoint.err.nonfatal",
450                                        accepted, ex);
451                     log.warn(msg, ex);
452                 }
453                 accepted = null;
454             }
455
456             if( ! running ) return null;
457             reinitializing = true;
458             // Restart endpoint when getting an IOException during accept
459 synchronized (threadSync) {
460                 if (reinitializing) {
461                     reinitializing = false;
462                     // 1) Attempt to close server socket
463 closeServerSocket();
464                     initialized = false;
465                     // 2) Reinit endpoint (recreate server socket)
466 try {
467                         msg = sm.getString("endpoint.warn.reinit");
468                         log.warn(msg);
469                         initEndpoint();
470                     } catch (Throwable t) {
471                         msg = sm.getString("endpoint.err.nonfatal",
472                                            serverSocket, t);
473                         log.error(msg, t);
474                     }
475                     // 3) If failed, attempt to restart endpoint
476 if (!initialized) {
477                         msg = sm.getString("endpoint.warn.restart");
478                         log.warn(msg);
479                         try {
480                             stopEndpoint();
481                             initEndpoint();
482                             startEndpoint();
483                         } catch (Throwable t) {
484                             msg = sm.getString("endpoint.err.fatal",
485                                                serverSocket, t);
486                             log.error(msg, t);
487                         }
488                         // Current thread is now invalid: kill it
489 throw new ThreadDeath ();
490                     }
491                 }
492             }
493
494         }
495
496         return accepted;
497     }
498
499     void setSocketOptions(Socket socket)
500         throws SocketException {
501         if(linger >= 0 )
502             socket.setSoLinger( true, linger);
503         if( tcpNoDelay )
504             socket.setTcpNoDelay(tcpNoDelay);
505         if( socketTimeout > 0 )
506             socket.setSoTimeout( socketTimeout );
507     }
508
509     
510     void processSocket(Socket s, TcpConnection con, Object [] threadData) {
511         // Process the connection
512 int step = 1;
513         try {
514             
515             // 1: Set socket options: timeout, linger, etc
516 setSocketOptions(s);
517             
518             // 2: SSL handshake
519 step = 2;
520             if (getServerSocketFactory() != null) {
521                 getServerSocketFactory().handshake(s);
522             }
523             
524             // 3: Process the connection
525 step = 3;
526             con.setEndpoint(this);
527             con.setSocket(s);
528             getConnectionHandler().processConnection(con, threadData);
529             
530         } catch (SocketException se) {
531             log.debug(sm.getString("endpoint.err.socket", s.getInetAddress()),
532                     se);
533             // Try to close the socket
534 try {
535                 s.close();
536             } catch (IOException e) {
537             }
538         } catch (Throwable t) {
539             if (step == 2) {
540                 if (log.isDebugEnabled()) {
541                     log.debug(sm.getString("endpoint.err.handshake"), t);
542                 }
543             } else {
544                 log.error(sm.getString("endpoint.err.unexpected"), t);
545             }
546             // Try to close the socket
547 try {
548                 s.close();
549             } catch (IOException e) {
550             }
551         } finally {
552             if (con != null) {
553                 con.recycle();
554             }
555         }
556     }
557     
558
559     // -------------------------------------------------- Master Slave Methods
560
561
562     /**
563      * Create (or allocate) and return an available processor for use in
564      * processing a specific HTTP request, if possible. If the maximum
565      * allowed processors have already been created and are in use, return
566      * <code>null</code> instead.
567      */
568     private MasterSlaveWorkerThread createWorkerThread() {
569
570         synchronized (workerThreads) {
571             if (workerThreads.size() > 0) {
572                 return ((MasterSlaveWorkerThread) workerThreads.pop());
573             }
574             if ((maxThreads > 0) && (curThreads < maxThreads)) {
575                 return (newWorkerThread());
576             } else {
577                 if (maxThreads < 0) {
578                     return (newWorkerThread());
579                 } else {
580                     return (null);
581                 }
582             }
583         }
584
585     }
586
587     
588     /**
589      * Create and return a new processor suitable for processing HTTP
590      * requests and returning the corresponding responses.
591      */
592     private MasterSlaveWorkerThread newWorkerThread() {
593
594         MasterSlaveWorkerThread workerThread =
595             new MasterSlaveWorkerThread(this, tp.getName() + "-" + (++curThreads));
596         workerThread.start();
597         created.addElement(workerThread);
598         return (workerThread);
599
600     }
601
602
603     /**
604      * Recycle the specified Processor so that it can be used again.
605      *
606      * @param processor The processor to be recycled
607      */
608     void recycleWorkerThread(MasterSlaveWorkerThread workerThread) {
609         workerThreads.push(workerThread);
610     }
611
612     
613     /**
614      * The background thread that listens for incoming TCP/IP connections and
615      * hands them off to an appropriate processor.
616      */
617     public void run() {
618
619         // Loop until we receive a shutdown command
620 while (running) {
621
622             // Loop if endpoint is paused
623 while (paused) {
624                 try {
625                     Thread.sleep(1000);
626                 } catch (InterruptedException e) {
627                     // Ignore
628 }
629             }
630
631             // Allocate a new worker thread
632 MasterSlaveWorkerThread workerThread = createWorkerThread();
633             if (workerThread == null) {
634                 try {
635                     // Wait a little for load to go down: as a result,
636 // no accept will be made until the concurrency is
637 // lower than the specified maxThreads, and current
638 // connections will wait for a little bit instead of
639 // failing right away.
640 Thread.sleep(100);
641                 } catch (InterruptedException e) {
642                     // Ignore
643 }
644                 continue;
645             }
646             
647             // Accept the next incoming connection from the server socket
648 Socket socket = acceptSocket();
649
650             // Hand this socket off to an appropriate processor
651 workerThread.assign(socket);
652
653             // The processor will recycle itself when it finishes
654
655         }
656
657         // Notify the threadStop() method that we have shut ourselves down
658 synchronized (threadSync) {
659             threadSync.notifyAll();
660         }
661
662     }
663
664
665     /**
666      * Start the background processing thread.
667      */
668     private void threadStart() {
669         thread = new Thread (this, tp.getName());
670         thread.setPriority(getThreadPriority());
671         thread.setDaemon(true);
672         thread.start();
673     }
674
675
676     /**
677      * Stop the background processing thread.
678      */
679     private void threadStop() {
680         thread = null;
681     }
682
683
684 }
685

A to Z: JavaDoc & Examples

---

Daily Java News & Articles

---

Open Source Projects

---

Open Source Codes

---

Free Computer Books

---

Remove Frame

---

Popular Tags