RAIDb2_LPRF


1   /**
2    * Sequoia: Database clustering technology.
3    * Copyright (C) 2002-2004 French National Institute For Research In Computer
4    * Science And Control (INRIA).
5    * Contact: sequoia@continuent.org
6    * 
7    * Licensed under the Apache License, Version 2.0 (the "License");
8    * you may not use this file except in compliance with the License.
9    * You may obtain a copy of the License at
10   * 
11   * http://www.apache.org/licenses/LICENSE-2.0
12   * 
13   * Unless required by applicable law or agreed to in writing, software
14   * distributed under the License is distributed on an "AS IS" BASIS,
15   * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
16   * See the License for the specific language governing permissions and
17   * limitations under the License. 
18   *
19   * Initial developer(s): Emmanuel Cecchet.
20   * Contributor(s): ______________________.
21   */
22  
23  package org.continuent.sequoia.controller.loadbalancer.raidb2;
24  
25  import java.sql.SQLException  ;
26  import java.util.ArrayList  ;
27  import java.util.Collection  ;
28  
29  import org.continuent.sequoia.common.exceptions.NoMoreBackendException;
30  import org.continuent.sequoia.common.exceptions.UnreachableBackendException;
31  import org.continuent.sequoia.common.i18n.Translate;
32  import org.continuent.sequoia.common.xml.DatabasesXmlTags;
33  import org.continuent.sequoia.controller.backend.DatabaseBackend;
34  import org.continuent.sequoia.controller.backend.result.ControllerResultSet;
35  import org.continuent.sequoia.controller.backend.result.ExecuteResult;
36  import org.continuent.sequoia.controller.cache.metadata.MetadataCache;
37  import org.continuent.sequoia.controller.loadbalancer.policies.WaitForCompletionPolicy;
38  import org.continuent.sequoia.controller.loadbalancer.policies.createtable.CreateTablePolicy;
39  import org.continuent.sequoia.controller.requests.SelectRequest;
40  import org.continuent.sequoia.controller.requests.StoredProcedure;
41  import org.continuent.sequoia.controller.virtualdatabase.VirtualDatabase;
42  
43  /**
44   * RAIDb-2 Round Robin load balancer featuring (Least Pending Requests First
45   * load balancing algorithm).
46   * <p>
47   * The read requests coming from the request manager are sent to the node that
48   * has the Least pending read requests among the nodes that can execute the
49   * request.
50   * 
51   * @author <a HREF="mailto:Emmanuel.Cecchet@inria.fr">Emmanuel Cecchet </a>
52   * @version 1.0
53   */
54  public class RAIDb2_LPRF extends RAIDb2
55  {
56  
57    /*
58     * How the code is organized ? 1. Member variables 2. Constructor(s) 3.
59     * Request handling 4. Debug/Monitoring
60     */
61  
62    /*
63     * Constructors
64     */
65  
66    /**
67     * Creates a new RAIDb-2 Round Robin request load balancer.
68     * 
69     * @param vdb the virtual database this load balancer belongs to.
70     * @param waitForCompletionPolicy how many backends must complete before
71     *          returning the result?
72     * @param createTablePolicy the policy defining how 'create table' statements
73     *          should be handled
74     * @exception Exception if an error occurs
75     */
76    public RAIDb2_LPRF(VirtualDatabase vdb,
77        WaitForCompletionPolicy waitForCompletionPolicy,
78        CreateTablePolicy createTablePolicy) throws Exception  
79    {
80      super(vdb, waitForCompletionPolicy, createTablePolicy);
81    }
82  
83    /*
84     * Request Handling
85     */
86  
87    /**
88     * Chooses the node to execute the request using a round-robin algorithm. If
89     * the next node has not the tables needed to execute the requests, we try the
90     * next one and so on until a suitable backend is found.
91     * 
92     * @param request an <code>SelectRequest</code>
93     * @param metadataCache cached metadata to use to construct the result set
94     * @return the corresponding <code>java.sql.ResultSet</code>
95     * @exception SQLException if an error occurs
96     * @see org.continuent.sequoia.controller.loadbalancer.raidb2.RAIDb2#statementExecuteQuery(SelectRequest,
97     *      MetadataCache)
98     */
99    public ControllerResultSet statementExecuteQuery(SelectRequest request,
100       MetadataCache metadataCache) throws SQLException  
101   {
102     // Choose a backend
103     try
104     {
105       vdb.acquireReadLockBackendLists();
106     }
107     catch (InterruptedException   e)
108     {
109       String   msg = Translate.get(
110           "loadbalancer.backendlist.acquire.readlock.failed", e);
111       logger.error(msg);
112       throw new SQLException  (msg);
113     }
114 
115     DatabaseBackend backend = null; // The backend that will execute the query
116 
117     // Note that vdb lock is released in the finally clause of this try/catch
118     // block
119     try
120     {
121       ArrayList   backends = vdb.getBackends();
122       int size = backends.size();
123 
124       if (size == 0)
125         throw new NoMoreBackendException(Translate.get(
126             "loadbalancer.execute.no.backend.available", request.getId()));
127 
128       // Choose the backend that has the least pending requests
129       int leastRequests = 0;
130       int enabledBackends = 0;
131       Collection   tables = request.getFrom();
132       if (tables == null)
133         throw new SQLException  (Translate.get("loadbalancer.from.not.found",
134             request.getSqlShortForm(vdb.getSqlShortFormLength())));
135 
136       for (int i = 0; i < size; i++)
137       {
138         DatabaseBackend b = (DatabaseBackend) backends.get(i);
139         if (b.isReadEnabled())
140         {
141           enabledBackends++;
142           if (b.hasTables(tables))
143           {
144             int pending = b.getPendingRequests().size();
145             if (((backend == null) || (pending < leastRequests)))
146             {
147               backend = b;
148               if (pending == 0)
149                 break; // Stop here we will never find a less loaded node
150               else
151                 leastRequests = pending;
152             }
153           }
154         }
155       }
156 
157       if (backend == null)
158       {
159         if (enabledBackends == 0)
160           throw new NoMoreBackendException(Translate.get(
161               "loadbalancer.execute.no.backend.enabled", request.getId()));
162         else
163           throw new SQLException  (Translate.get(
164               "loadbalancer.backend.no.required.tables", tables.toString()));
165       }
166 
167     }
168     catch (RuntimeException   e)
169     {
170       String   msg = Translate.get("loadbalancer.request.failed.on.backend",
171           new String  []{request.getSqlShortForm(vdb.getSqlShortFormLength()),
172               backend.getName(), e.getMessage()});
173       logger.error(msg, e);
174       throw new SQLException  (msg);
175     }
176     finally
177     {
178       vdb.releaseReadLockBackendLists();
179     }
180 
181     // Execute the request on the chosen backend
182     ControllerResultSet rs = null;
183     try
184     {
185       rs = executeReadRequestOnBackend(request, backend, metadataCache);
186     }
187     catch (UnreachableBackendException se)
188     {
189       // Try on another backend
190       return statementExecuteQuery(request, metadataCache);
191     }
192     catch (SQLException   se)
193     {
194       String   msg = Translate.get("loadbalancer.request.failed", new String  []{
195           String.valueOf(request.getId()), se.getMessage()});
196       if (logger.isInfoEnabled())
197         logger.info(msg);
198       throw new SQLException  (msg);
199     }
200     catch (Throwable   e)
201     {
202       String   msg = Translate.get("loadbalancer.request.failed.on.backend",
203           new String  []{request.getSqlShortForm(vdb.getSqlShortFormLength()),
204               backend.getName(), e.getMessage()});
205       logger.error(msg, e);
206       throw new SQLException  (msg);
207     }
208 
209     return rs;
210   }
211 
212   /**
213    * Chooses the node to execute the stored procedure using a LPRF algorithm. If
214    * the next node has not the needed stored procedure, we try the next one and
215    * so on until a suitable backend is found.
216    * 
217    * @see org.continuent.sequoia.controller.loadbalancer.AbstractLoadBalancer#readOnlyCallableStatementExecuteQuery(StoredProcedure,
218    *      MetadataCache)
219    */
220   public ControllerResultSet readOnlyCallableStatementExecuteQuery(
221       StoredProcedure proc, MetadataCache metadataCache) throws SQLException  
222   {
223     return (ControllerResultSet) callStoredProcedure(proc,
224         CALLABLE_STATEMENT_EXECUTE_QUERY, metadataCache);
225   }
226 
227   /**
228    * Chooses the node to execute the stored procedure using a LPRF algorithm. If
229    * the next node has not the needed stored procedure, we try the next one and
230    * so on until a suitable backend is found.
231    * 
232    * @see org.continuent.sequoia.controller.loadbalancer.AbstractLoadBalancer#readOnlyCallableStatementExecute(org.continuent.sequoia.controller.requests.StoredProcedure,
233    *      org.continuent.sequoia.controller.cache.metadata.MetadataCache)
234    */
235   public ExecuteResult readOnlyCallableStatementExecute(StoredProcedure proc,
236       MetadataCache metadataCache) throws SQLException  
237   {
238     return (ExecuteResult) callStoredProcedure(proc,
239         CALLABLE_STATEMENT_EXECUTE, metadataCache);
240   }
241 
242   /**
243    * Common code to execute a StoredProcedure using executeQuery or
244    * executeUpdate on a backend chosen using a LPRF algorithm.
245    * 
246    * @param proc a <code>StoredProcedure</code>
247    * @param callType one of CALLABLE_STATEMENT_EXECUTE_QUERY or
248    *          CALLABLE_STATEMENT_EXECUTE
249    * @param metadataCache the metadataCache if any or null
250    * @return a <code>ControllerResultSet</code> or an
251    *         <code>ExecuteResult</code> object
252    * @throws SQLException if an error occurs
253    */
254   private Object   callStoredProcedure(StoredProcedure proc, int callType,
255       MetadataCache metadataCache) throws SQLException  
256   {
257     // Choose a backend
258     try
259     {
260       vdb.acquireReadLockBackendLists();
261     }
262     catch (InterruptedException   e)
263     {
264       String   msg = Translate.get(
265           "loadbalancer.backendlist.acquire.readlock.failed", e);
266       logger.error(msg);
267       throw new SQLException  (msg);
268     }
269 
270     DatabaseBackend backend = null; // The backend that will execute the query
271 
272     // Note that vdb lock is released in the finally clause of this try/catch
273     // block
274     try
275     {
276       DatabaseBackend failedBackend = null;
277       SQLException   failedException = null;
278       Object   result = null;
279       do
280       {
281         ArrayList   backends = vdb.getBackends();
282         int size = backends.size();
283 
284         if (size == 0)
285           throw new NoMoreBackendException(Translate.get(
286               "loadbalancer.execute.no.backend.available", proc.getId()));
287 
288         // Choose the backend that has the least pending requests
289         int leastRequests = 0;
290         int enabledBackends = 0;
291 
292         for (int i = 0; i < size; i++)
293         {
294           DatabaseBackend b = (DatabaseBackend) backends.get(i);
295           if (b.isReadEnabled())
296           {
297             enabledBackends++;
298             if (b.hasStoredProcedure(proc.getProcedureKey(), proc
299                 .getNbOfParameters()))
300             {
301               int pending = b.getPendingRequests().size();
302               if ((b != failedBackend)
303                   && ((backend == null) || (pending < leastRequests)))
304               {
305                 backend = b;
306                 if (pending == 0)
307                   break; // Stop here we will never find a less loaded node
308                 else
309                   leastRequests = pending;
310               }
311             }
312           }
313         }
314 
315         if (backend == null)
316         {
317           if (enabledBackends == 0)
318             throw new SQLException  (Translate
319                 .get("loadbalancer.storedprocedure.backend.no.enabled", proc
320                     .getId()));
321           else if (failedBackend == null)
322             throw new SQLException  (Translate.get(
323                 "loadbalancer.backend.no.required.storedprocedure", proc
324                     .getProcedureKey()));
325           else
326             // Bad query, the only backend that could execute it has failed
327             throw failedException;
328         }
329 
330         // Execute the request on the chosen backend
331         boolean toDisable = false;
332         try
333         {
334           switch (callType)
335           {
336             case CALLABLE_STATEMENT_EXECUTE_QUERY :
337               result = executeStoredProcedureOnBackend(proc, true, backend,
338                   metadataCache);
339               break;
340             case CALLABLE_STATEMENT_EXECUTE :
341               result = executeStoredProcedureOnBackend(proc, false, backend,
342                   metadataCache);
343               break;
344             default :
345               throw new RuntimeException  ("Unhandled call type " + callType
346                   + " in executeLPRF");
347           }
348           if (failedBackend != null)
349           { // Previous backend failed
350             if (logger.isWarnEnabled())
351               logger.warn(Translate.get("loadbalancer.storedprocedure.status",
352                   new String  []{String.valueOf(proc.getId()), backend.getName(),
353                       failedBackend.getName()}));
354             toDisable = true;
355           }
356         }
357         catch (UnreachableBackendException se)
358         {
359           // Retry on an other backend.
360           continue;
361         }
362         catch (SQLException   se)
363         {
364           if (failedBackend != null)
365           { // Bad query, no backend can execute it
366             String   msg = Translate.get(
367                 "loadbalancer.storedprocedure.failed.twice", new String  []{
368                     String.valueOf(proc.getId()), se.getMessage()});
369             if (logger.isInfoEnabled())
370               logger.info(msg);
371             throw new SQLException  (msg);
372           }
373           else
374           { // We are the first to fail on this query
375             failedBackend = backend;
376             failedException = se;
377             if (logger.isInfoEnabled())
378               logger.info(Translate.get(
379                   "loadbalancer.storedprocedure.failed.on.backend",
380                   new String  []{
381                       proc.getSqlShortForm(vdb.getSqlShortFormLength()),
382                       backend.getName(), se.getMessage()}));
383             continue;
384           }
385         }
386 
387         if (toDisable)
388         { // retry has succeeded and we need to disable the first node that
389           // failed
390           try
391           {
392             if (logger.isWarnEnabled())
393               logger.warn(Translate.get("loadbalancer.backend.disabling",
394                   failedBackend.getName()));
395             disableBackend(failedBackend, true);
396           }
397           catch (SQLException   ignore)
398           {
399           }
400           finally
401           {
402             failedBackend = null; // to exit the do{}while
403           }
404         }
405       }
406       while (failedBackend != null);
407       return result;
408     }
409     catch (RuntimeException   e)
410     {
411       String   msg = Translate.get(
412           "loadbalancer.storedprocedure.failed.on.backend", new String  []{
413               proc.getSqlShortForm(vdb.getSqlShortFormLength()),
414               backend.getName(), e.getMessage()});
415       logger.fatal(msg, e);
416       throw new SQLException  (msg);
417     }
418     finally
419     {
420       vdb.releaseReadLockBackendLists();
421     }
422   }
423 
424   /*
425    * Debug/Monitoring
426    */
427 
428   /**
429    * Gets information about the request load balancer.
430    * 
431    * @return <code>String</code> containing information
432    */
433   public String   getInformation()
434   {
435     // We don't lock since we don't need a completely accurate value
436     int size = vdb.getBackends().size();
437 
438     if (size == 0)
439       return "RAIDb-2 Least Pending Requests First load balancer: !!!Warning!!! No backend nodes found\n";
440     else
441       return "RAIDb-2 Least Pending Requests First load balancer (" + size
442           + " backends)\n";
443   }
444 
445   /**
446    * @see org.continuent.sequoia.controller.loadbalancer.raidb2.RAIDb2#getRaidb2Xml
447    */
448   public String   getRaidb2Xml()
449   {
450     return "<" + DatabasesXmlTags.ELT_RAIDb_2_LeastPendingRequestsFirst + "/>";
451   }
452 }
A to Z: JavaDoc & Examples Daily Java News & Articles Open Source Projects Open Source Codes Free Computer Books Remove Frame
Popular Tags