View Javadoc

1   //
2   //  ========================================================================
3   //  Copyright (c) 1995-2014 Mort Bay Consulting Pty. Ltd.
4   //  ------------------------------------------------------------------------
5   //  All rights reserved. This program and the accompanying materials
6   //  are made available under the terms of the Eclipse Public License v1.0
7   //  and Apache License v2.0 which accompanies this distribution.
8   //
9   //      The Eclipse Public License is available at
10  //      http://www.eclipse.org/legal/epl-v10.html
11  //
12  //      The Apache License v2.0 is available at
13  //      http://www.opensource.org/licenses/apache2.0.php
14  //
15  //  You may elect to redistribute this code under either of these licenses.
16  //  ========================================================================
17  //
18  
19  package org.eclipse.jetty.servlets;
20  
21  import java.io.IOException;
22  import java.util.Queue;
23  import java.util.concurrent.ConcurrentLinkedQueue;
24  import java.util.concurrent.Semaphore;
25  import java.util.concurrent.TimeUnit;
26  
27  import javax.servlet.Filter;
28  import javax.servlet.FilterChain;
29  import javax.servlet.FilterConfig;
30  import javax.servlet.ServletContext;
31  import javax.servlet.ServletException;
32  import javax.servlet.ServletRequest;
33  import javax.servlet.ServletResponse;
34  import javax.servlet.http.HttpServletRequest;
35  import javax.servlet.http.HttpServletResponse;
36  import javax.servlet.http.HttpSession;
37  
38  import org.eclipse.jetty.continuation.Continuation;
39  import org.eclipse.jetty.continuation.ContinuationListener;
40  import org.eclipse.jetty.continuation.ContinuationSupport;
41  import org.eclipse.jetty.server.handler.ContextHandler;
42  
43  /**
44   * Quality of Service Filter.
45   * 
46   * This filter limits the number of active requests to the number set by the "maxRequests" init parameter (default 10).
47   * If more requests are received, they are suspended and placed on priority queues.  Priorities are determined by 
48   * the {@link #getPriority(ServletRequest)} method and are a value between 0 and the value given by the "maxPriority" 
49   * init parameter (default 10), with higher values having higher priority.
50   * </p><p>
51   * This filter is ideal to prevent wasting threads waiting for slow/limited 
52   * resources such as a JDBC connection pool.  It avoids the situation where all of a 
53   * containers thread pool may be consumed blocking on such a slow resource.
54   * By limiting the number of active threads, a smaller thread pool may be used as 
55   * the threads are not wasted waiting.  Thus more memory may be available for use by 
56   * the active threads.
57   * </p><p>
58   * Furthermore, this filter uses a priority when resuming waiting requests. So that if
59   * a container is under load, and there are many requests waiting for resources,
60   * the {@link #getPriority(ServletRequest)} method is used, so that more important 
61   * requests are serviced first.     For example, this filter could be deployed with a 
62   * maxRequest limit slightly smaller than the containers thread pool and a high priority 
63   * allocated to admin users.  Thus regardless of load, admin users would always be
64   * able to access the web application.
65   * </p><p>
66   * The maxRequest limit is policed by a {@link Semaphore} and the filter will wait a short while attempting to acquire
67   * the semaphore. This wait is controlled by the "waitMs" init parameter and allows the expense of a suspend to be
68   * avoided if the semaphore is shortly available.  If the semaphore cannot be obtained, the request will be suspended
69   * for the default suspend period of the container or the valued set as the "suspendMs" init parameter.
70   * </p><p>
71   * If the "managedAttr" init parameter is set to true, then this servlet is set as a {@link ServletContext} attribute with the 
72   * filter name as the attribute name.  This allows context external mechanism (eg JMX via {@link ContextHandler#MANAGED_ATTRIBUTES}) to
73   * manage the configuration of the filter.
74   * </p>
75   * 
76   *
77   */
78  public class QoSFilter implements Filter
79  {
80      final static int __DEFAULT_MAX_PRIORITY=10;
81      final static int __DEFAULT_PASSES=10;
82      final static int __DEFAULT_WAIT_MS=50;
83      final static long __DEFAULT_TIMEOUT_MS = -1;
84      
85      final static String MANAGED_ATTR_INIT_PARAM="managedAttr";
86      final static String MAX_REQUESTS_INIT_PARAM="maxRequests";
87      final static String MAX_PRIORITY_INIT_PARAM="maxPriority";
88      final static String MAX_WAIT_INIT_PARAM="waitMs";
89      final static String SUSPEND_INIT_PARAM="suspendMs";
90      
91      ServletContext _context;
92  
93      protected long _waitMs;
94      protected long _suspendMs;
95      protected int _maxRequests;
96      
97      private Semaphore _passes;
98      private Queue<Continuation>[] _queue;
99      private ContinuationListener[] _listener;
100     private String _suspended="QoSFilter@"+this.hashCode();
101     
102     /* ------------------------------------------------------------ */
103     /**
104      * @see javax.servlet.Filter#init(javax.servlet.FilterConfig)
105      */
106     public void init(FilterConfig filterConfig) 
107     {
108         _context=filterConfig.getServletContext();
109 
110         int max_priority=__DEFAULT_MAX_PRIORITY;
111         if (filterConfig.getInitParameter(MAX_PRIORITY_INIT_PARAM)!=null)
112             max_priority=Integer.parseInt(filterConfig.getInitParameter(MAX_PRIORITY_INIT_PARAM));
113         _queue=new Queue[max_priority+1];
114         _listener = new ContinuationListener[max_priority + 1];
115         for (int p=0;p<_queue.length;p++)
116         {
117             _queue[p]=new ConcurrentLinkedQueue<Continuation>();
118 
119             final int priority=p;
120             _listener[p] = new ContinuationListener()
121             {
122                 public void onComplete(Continuation continuation)
123                 {}
124 
125                 public void onTimeout(Continuation continuation)
126                 {
127                     _queue[priority].remove(continuation);
128                 }
129             };
130         }
131         
132         int maxRequests=__DEFAULT_PASSES;
133         if (filterConfig.getInitParameter(MAX_REQUESTS_INIT_PARAM)!=null)
134             maxRequests=Integer.parseInt(filterConfig.getInitParameter(MAX_REQUESTS_INIT_PARAM));
135         _passes=new Semaphore(maxRequests,true);
136         _maxRequests = maxRequests;
137         
138         long wait = __DEFAULT_WAIT_MS;
139         if (filterConfig.getInitParameter(MAX_WAIT_INIT_PARAM)!=null)
140             wait=Integer.parseInt(filterConfig.getInitParameter(MAX_WAIT_INIT_PARAM));
141         _waitMs=wait;
142         
143         long suspend = __DEFAULT_TIMEOUT_MS;
144         if (filterConfig.getInitParameter(SUSPEND_INIT_PARAM)!=null)
145             suspend=Integer.parseInt(filterConfig.getInitParameter(SUSPEND_INIT_PARAM));
146         _suspendMs=suspend;
147 
148         if (_context!=null && Boolean.parseBoolean(filterConfig.getInitParameter(MANAGED_ATTR_INIT_PARAM)))
149             _context.setAttribute(filterConfig.getFilterName(),this);
150     }
151     
152     /* ------------------------------------------------------------ */
153     /**
154      * @see javax.servlet.Filter#doFilter(javax.servlet.ServletRequest, javax.servlet.ServletResponse, javax.servlet.FilterChain)
155      */
156     public void doFilter(ServletRequest request, ServletResponse response, FilterChain chain) 
157     throws IOException, ServletException
158     {
159         boolean accepted=false;
160         try
161         {
162             if (request.getAttribute(_suspended)==null)
163             {
164                 accepted=_passes.tryAcquire(_waitMs,TimeUnit.MILLISECONDS);
165                 if (accepted)
166                 {
167                     request.setAttribute(_suspended,Boolean.FALSE);
168                 }
169                 else
170                 {
171                     request.setAttribute(_suspended,Boolean.TRUE);
172                     int priority = getPriority(request);
173                     Continuation continuation = ContinuationSupport.getContinuation(request);
174                     if (_suspendMs>0)
175                         continuation.setTimeout(_suspendMs);
176                     continuation.suspend();
177                     continuation.addContinuationListener(_listener[priority]);
178                     _queue[priority].add(continuation);
179                     return;
180                 }
181             }
182             else
183             {
184                 Boolean suspended=(Boolean)request.getAttribute(_suspended);
185                 
186                 if (suspended.booleanValue())
187                 {
188                     request.setAttribute(_suspended,Boolean.FALSE);
189                     if (request.getAttribute("javax.servlet.resumed")==Boolean.TRUE)
190                     {
191                         _passes.acquire();
192                         accepted=true;
193                     }
194                     else 
195                     {
196                         // Timeout! try 1 more time.
197                         accepted = _passes.tryAcquire(_waitMs,TimeUnit.MILLISECONDS);
198                     }
199                 }
200                 else
201                 {
202                     // pass through resume of previously accepted request
203                     _passes.acquire();
204                     accepted = true;
205                 }
206             }
207 
208             if (accepted)
209             {
210                 chain.doFilter(request,response);
211             }
212             else
213             {
214                 ((HttpServletResponse)response).sendError(HttpServletResponse.SC_SERVICE_UNAVAILABLE);
215             }
216         }
217         catch(InterruptedException e)
218         {
219             _context.log("QoS",e);
220             ((HttpServletResponse)response).sendError(HttpServletResponse.SC_SERVICE_UNAVAILABLE);
221         }
222         finally
223         {
224             if (accepted)
225             {
226                 for (int p=_queue.length;p-->0;)
227                 {
228                     Continuation continutaion=_queue[p].poll();
229                     if (continutaion!=null && continutaion.isSuspended())
230                     {
231                         continutaion.resume();
232                         break;
233                     }
234                 }
235                 _passes.release();
236             }
237         }
238     }
239 
240     /** 
241      * Get the request Priority.
242      * <p> The default implementation assigns the following priorities:<ul>
243      * <li> 2 - for a authenticated request
244      * <li> 1 - for a request with valid /non new session 
245      * <li> 0 - for all other requests.
246      * </ul>
247      * This method may be specialised to provide application specific priorities.
248      * 
249      * @param request
250      * @return the request priority
251      */
252     protected int getPriority(ServletRequest request)
253     {
254         HttpServletRequest baseRequest = (HttpServletRequest)request;
255         if (baseRequest.getUserPrincipal() != null )
256             return 2;
257         else 
258         {
259             HttpSession session = baseRequest.getSession(false);
260             if (session!=null && !session.isNew()) 
261                 return 1;
262             else
263                 return 0;
264         }
265     }
266 
267 
268     /* ------------------------------------------------------------ */
269     /**
270      * @see javax.servlet.Filter#destroy()
271      */
272     public void destroy(){}
273 
274     /* ------------------------------------------------------------ */
275     /** 
276      * Get the (short) amount of time (in milliseconds) that the filter would wait
277      * for the semaphore to become available before suspending a request.
278      * 
279      * @return wait time (in milliseconds)
280      */
281     public long getWaitMs()
282     {
283         return _waitMs;
284     }
285 
286     /* ------------------------------------------------------------ */
287     /**
288      * Set the (short) amount of time (in milliseconds) that the filter would wait
289      * for the semaphore to become available before suspending a request.
290      * 
291      * @param value wait time (in milliseconds)
292      */
293     public void setWaitMs(long value)
294     {
295         _waitMs = value;
296     }
297 
298     /* ------------------------------------------------------------ */
299     /**
300      * Get the amount of time (in milliseconds) that the filter would suspend
301      * a request for while waiting for the semaphore to become available.
302      * 
303      * @return suspend time (in milliseconds)
304      */
305     public long getSuspendMs()
306     {
307         return _suspendMs;
308     }
309 
310     /* ------------------------------------------------------------ */
311     /**
312      * Set the amount of time (in milliseconds) that the filter would suspend
313      * a request for while waiting for the semaphore to become available.
314      * 
315      * @param value suspend time (in milliseconds)
316      */
317     public void setSuspendMs(long value)
318     {
319         _suspendMs = value;
320     }
321 
322     /* ------------------------------------------------------------ */
323     /**
324      * Get the maximum number of requests allowed to be processed
325      * at the same time.
326      * 
327      * @return maximum number of requests
328      */
329     public int getMaxRequests()
330     {
331         return _maxRequests;
332     }
333 
334     /* ------------------------------------------------------------ */
335     /**
336      * Set the maximum number of requests allowed to be processed
337      * at the same time.
338      * 
339      * @param value the number of requests
340      */
341     public void setMaxRequests(int value)
342     {
343         _passes = new Semaphore((value-_maxRequests+_passes.availablePermits()), true);
344         _maxRequests = value;
345     }
346 
347 }