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