Java > Open Source Codes > org > continuent > sequoia > controller > loadbalancer > AbstractLoadBalancer


1 /**
2  * Sequoia: Database clustering technology.
3  * Copyright (C) 2002-2004 French National Institute For Research In Computer
4  * Science And Control (INRIA).
5  * Copyright (C) 2005 AmicoSoft, Inc. dba Emic Networks
6  * Copyright (C) 2005-2006 Continuent, Inc.
7  * Contact: sequoia@continuent.org
8  *
9  * Licensed under the Apache License, Version 2.0 (the "License");
10  * you may not use this file except in compliance with the License.
11  * You may obtain a copy of the License at
12  *
13  * http://www.apache.org/licenses/LICENSE-2.0
14  *
15  * Unless required by applicable law or agreed to in writing, software
16  * distributed under the License is distributed on an "AS IS" BASIS,
17  * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
18  * See the License for the specific language governing permissions and
19  * limitations under the License.
20  *
21  * Initial developer(s): Emmanuel Cecchet.
22  * Contributor(s): Vadim Kassin, Jaco Swart, Jean-Bernard van Zuylen
23  */
24
25 package org.continuent.sequoia.controller.loadbalancer;
26
27 import java.sql.CallableStatement ;
28 import java.sql.Connection ;
29 import java.sql.PreparedStatement ;
30 import java.sql.ResultSet ;
31 import java.sql.SQLException ;
32 import java.sql.SQLWarning ;
33 import java.sql.Statement ;
34 import java.util.ArrayList ;
35 import java.util.Iterator ;
36 import java.util.LinkedList ;
37 import java.util.List ;
38
39 import org.continuent.sequoia.common.exceptions.BadConnectionException;
40 import org.continuent.sequoia.common.exceptions.NoMoreBackendException;
41 import org.continuent.sequoia.common.exceptions.UnreachableBackendException;
42 import org.continuent.sequoia.common.i18n.Translate;
43 import org.continuent.sequoia.common.log.Trace;
44 import org.continuent.sequoia.common.xml.DatabasesXmlTags;
45 import org.continuent.sequoia.common.xml.XmlComponent;
46 import org.continuent.sequoia.controller.backend.DatabaseBackend;
47 import org.continuent.sequoia.controller.backend.DriverCompliance;
48 import org.continuent.sequoia.controller.backend.result.ControllerResultSet;
49 import org.continuent.sequoia.controller.backend.result.ExecuteResult;
50 import org.continuent.sequoia.controller.backend.result.ExecuteUpdateResult;
51 import org.continuent.sequoia.controller.backend.result.GeneratedKeysResult;
52 import org.continuent.sequoia.controller.cache.metadata.MetadataCache;
53 import org.continuent.sequoia.controller.connection.AbstractConnectionManager;
54 import org.continuent.sequoia.controller.connection.PooledConnection;
55 import org.continuent.sequoia.controller.core.ControllerConstants;
56 import org.continuent.sequoia.controller.loadbalancer.policies.WaitForCompletionPolicy;
57 import org.continuent.sequoia.controller.loadbalancer.tasks.AbstractTask;
58 import org.continuent.sequoia.controller.locks.ReadPrioritaryFIFOWriteLock;
59 import org.continuent.sequoia.controller.protocol.PreparedStatementSerialization;
60 import org.continuent.sequoia.controller.recoverylog.RecoveryLog;
61 import org.continuent.sequoia.controller.requestmanager.TransactionMetaData;
62 import org.continuent.sequoia.controller.requests.AbstractRequest;
63 import org.continuent.sequoia.controller.requests.AbstractWriteRequest;
64 import org.continuent.sequoia.controller.requests.SelectRequest;
65 import org.continuent.sequoia.controller.requests.StoredProcedure;
66 import org.continuent.sequoia.controller.sql.macros.MacrosHandler;
67 import org.continuent.sequoia.controller.virtualdatabase.VirtualDatabase;
68 import org.continuent.sequoia.controller.virtualdatabase.protocol.SuspendWritesMessage;
69
70 /**
71  * The Request Load Balancer should implement the load balancing of the requests
72  * among the backend nodes.
73  * <p>
74  * The requests comes from the Request Controller and are sent to the Connection
75  * Managers.
76  *
77  * @author <a HREF="mailto:Emmanuel.Cecchet@inria.fr">Emmanuel Cecchet </a>
78  * @author <a HREF="mailto:vadim@kase.kz">Vadim Kassin </a>
79  * @author <a HREF="mailto:jaco.swart@iblocks.co.uk">Jaco Swart </a>
80  * @author <a HREF="mailto:jbvanzuylen@transwide.com">Jean-Bernard van Zuylen
81  * </a>
82  * @version 1.0
83  */
84 public abstract class AbstractLoadBalancer implements XmlComponent
85 {
86
87   //
88 // How the code is organized ?
89 //
90 // 1. Member variables/Constructor
91 // 2. Getter/Setter (possibly in alphabetical order)
92 // 3. Request handling
93 // 4. Transaction management
94 // 5. Backend management
95 // 6. Debug/Monitoring
96 //
97
98   // Virtual Database this load balancer is attached to.
99 protected VirtualDatabase vdb;
100   protected RecoveryLog recoveryLog;
101   protected int raidbLevel;
102   protected int parsingGranularity;
103   /** Reference to distributed virtual database total order queue */
104   protected LinkedList totalOrderQueue;
105
106   protected MacrosHandler macroHandler;
107
108   /**
109    * List of enabled backends (includes backends in either ENABLED or DISABLING
110    * state).
111    *
112    * @see org.continuent.sequoia.common.jmx.management.BackendState
113    */
114   protected ArrayList enabledBackends;
115   protected ReadPrioritaryFIFOWriteLock backendListLock = new ReadPrioritaryFIFOWriteLock();
116
117   /** Should we wait for all backends to commit before returning ? */
118   public WaitForCompletionPolicy waitForCompletionPolicy;
119
120   private static int defaultTransactionIsolationLevel;
121
122   protected static Trace logger = Trace
123                                                             .getLogger("org.continuent.sequoia.controller.loadbalancer");
124
125   /**
126    * Generic constructor that sets some member variables and checks that
127    * backends are in the disabled state
128    *
129    * @param vdb The virtual database this load balancer belongs to
130    * @param raidbLevel The RAIDb level of this load balancer
131    * @param parsingGranularity The parsing granularity needed by this load
132    * balancer
133    */
134   protected AbstractLoadBalancer(VirtualDatabase vdb, int raidbLevel,
135       int parsingGranularity) throws SQLException
136   {
137     this.raidbLevel = raidbLevel;
138     this.parsingGranularity = parsingGranularity;
139     this.vdb = vdb;
140     this.totalOrderQueue = vdb.getTotalOrderQueue();
141     this.enabledBackends = new ArrayList ();
142     try
143     {
144       vdb.acquireReadLockBackendLists();
145     }
146     catch (InterruptedException e)
147     {
148       String msg = Translate.get(
149           "loadbalancer.backendlist.acquire.readlock.failed", e);
150       logger.error(msg);
151       throw new SQLException (msg);
152     }
153     int size = vdb.getBackends().size();
154     ArrayList backends = vdb.getBackends();
155     for (int i = 0; i < size; i++)
156     {
157       DatabaseBackend backend = (DatabaseBackend) backends.get(i);
158       if (backend.isReadEnabled() || backend.isWriteEnabled())
159       {
160         if (logger.isWarnEnabled())
161           logger.warn(Translate.get(
162               "loadbalancer.constructor.backends.not.disabled", backend
163                   .getName()));
164         try
165         {
166           disableBackend(backend, true);
167         }
168         catch (Exception e)
169         { // Set the disabled state anyway
170 backend.disable();
171         }
172       }
173     }
174     vdb.releaseReadLockBackendLists();
175   }
176
177   //
178 // Getter/Setter methods
179 //
180
181   /**
182    * Returns the defaultTransactionIsolationLevel value.
183    *
184    * @return Returns the defaultTransactionIsolationLevel.
185    */
186   public static final int getDefaultTransactionIsolationLevel()
187   {
188     return defaultTransactionIsolationLevel;
189   }
190
191   /**
192    * Sets the defaultTransactionIsolationLevel value.
193    *
194    * @param defaultTransactionIsolationLevel The
195    * defaultTransactionIsolationLevel to set.
196    */
197   public final void setDefaultTransactionIsolationLevel(
198       int defaultTransactionIsolationLevel)
199   {
200     AbstractLoadBalancer.defaultTransactionIsolationLevel = defaultTransactionIsolationLevel;
201   }
202
203   /**
204    * This sets the macro handler for this load balancer. Handling macros
205    * prevents different backends to generate different values when interpreting
206    * the macros which could result in data inconsitencies.
207    *
208    * @param handler <code>MacrosHandler</code> instance
209    */
210   public void setMacroHandler(MacrosHandler handler)
211   {
212     this.macroHandler = handler;
213   }
214
215   /**
216    * Get the needed query parsing granularity.
217    *
218    * @return needed query parsing granularity
219    */
220   public int getParsingGranularity()
221   {
222     return parsingGranularity;
223   }
224
225   /**
226    * Returns the RAIDbLevel.
227    *
228    * @return int the RAIDb level
229    */
230   public int getRAIDbLevel()
231   {
232     return raidbLevel;
233   }
234
235   /**
236    * Returns the recoveryLog value.
237    *
238    * @return Returns the recoveryLog.
239    */
240   public final RecoveryLog getRecoveryLog()
241   {
242     return recoveryLog;
243   }
244
245   /**
246    * Sets the recoveryLog value.
247    *
248    * @param recoveryLog The recoveryLog to set.
249    */
250   public final void setRecoveryLog(RecoveryLog recoveryLog)
251   {
252     this.recoveryLog = recoveryLog;
253   }
254
255   /**
256    * Associate a weight to a backend identified by its logical name.
257    *
258    * @param name the backend name
259    * @param w the weight
260    * @throws SQLException if an error occurs
261    */
262   public void setWeight(String name, int w) throws SQLException
263   {
264     throw new SQLException ("Weight is not supported by this load balancer");
265   }
266
267   //
268 // Utility functions
269 //
270
271   /**
272    * Acquire the given lock and check the number of threads. Throw a
273    * NoMoreBackendException if no thread is available else returns the number of
274    * threads.
275    *
276    * @param request object to remove from the total order queue in case no
277    * backend is available
278    * @param requestDescription description of the request to put in the error
279    * message in case of an error
280    * @return the number of threads in the acquired list
281    * @throws SQLException if there was a problem to acquire the lock on the
282    * enabled backend list
283    * @throws NoMoreBackendException if no backends are available anymore
284    */
285   protected int acquireLockAndCheckNbOfThreads(Object request,
286       String requestDescription) throws SQLException , NoMoreBackendException
287   {
288     try
289     {
290       backendListLock.acquireRead();
291     }
292     catch (InterruptedException e)
293     {
294       String msg = Translate.get(
295           "loadbalancer.backendlist.acquire.readlock.failed", e);
296       logger.error(msg);
297       throw new SQLException (msg);
298     }
299
300     int nbOfThreads = enabledBackends.size();
301     if (nbOfThreads == 0)
302     {
303       releaseLockAndUnlockNextQuery(request);
304       throw new NoMoreBackendException(Translate
305           .get("loadbalancer.backendlist.empty"));
306     }
307     else
308     {
309       if (logger.isDebugEnabled())
310         logger.debug(Translate.get("loadbalancer.execute.on.several",
311             new String []{requestDescription, String.valueOf(nbOfThreads)}));
312     }
313     return nbOfThreads;
314   }
315
316   /**
317    * Returns the number of nodes to wait for according to the defined
318    * <code>waitForCompletion</code> policy.
319    *
320    * @param nbOfThreads total number of threads
321    * @return int number of threads to wait for
322    */
323   protected int getNbToWait(int nbOfThreads)
324   {
325     int nbToWait;
326     switch (waitForCompletionPolicy.getPolicy())
327     {
328       case WaitForCompletionPolicy.FIRST :
329         nbToWait = 1;
330         break;
331       case WaitForCompletionPolicy.MAJORITY :
332         nbToWait = nbOfThreads / 2 + 1;
333         break;
334       case WaitForCompletionPolicy.ALL :
335         nbToWait = nbOfThreads;
336         break;
337       default :
338         logger
339             .warn(Translate.get("loadbalancer.waitforcompletion.unsupported"));
340         nbToWait = nbOfThreads;
341         break;
342     }
343     return nbToWait;
344   }
345
346   /**
347    * Interprets the macros in the request (depending on the
348    * <code>MacroHandler</code> set for this class) and modify either the
349    * skeleton or the query itself. Note that the given object is directly
350    * modified.
351    *
352    * @param request the request to process
353    */
354   public void handleMacros(AbstractRequest request)
355   {
356     if (macroHandler == null)
357       return;
358
359     // Do not handle macros for requests that don't need it.
360 if (!request.needsMacroProcessing())
361       return;
362
363     macroHandler.processMacros(request);
364   }
365
366   /**
367    * Release the backend list lock and remove the current query from the head of
368    * the total order queue to unlock the next query.
369    *
370    * @param currentQuery the current query to remove from the total order queue
371    */
372   protected void releaseLockAndUnlockNextQuery(Object currentQuery)
373   {
374     backendListLock.releaseRead();
375
376     // Unblock next query from total order queue
377 removeObjectFromAndNotifyTotalOrderQueue(currentQuery);
378   }
379
380   /**
381    * Remove an entry of the total order queue (usually the head) and notify the
382    * queue so that the next queries can be scheduled.
383    *
384    * @param request Object that should be removed from the total order queue
385    */
386   public void removeObjectFromAndNotifyTotalOrderQueue(Object request)
387   {
388     if ((totalOrderQueue != null) && (request != null))
389     {
390       synchronized (totalOrderQueue)
391       {
392         try
393         {
394           if (totalOrderQueue.remove(request))
395           {
396             if (logger.isDebugEnabled())
397               logger.debug("Removed " + request + " from total order queue");
398             totalOrderQueue.notifyAll();
399           }
400           else if (logger.isDebugEnabled())
401           {
402             logger.debug(request + " was not in the total order queue");
403           }
404         }
405         catch (RuntimeException e)
406         {
407           logger.warn("Unable to remove request " + request
408               + " from total order queue", e);
409         }
410       }
411     }
412   }
413
414   /**
415    * Wait for the completion of the given task. Note that this method must be
416    * called within a synchronized block on the task.
417    *
418    * @param timeout timeout in ms for this task
419    * @param requestDescription description of the request to put in the error
420    * message in case of a timeout
421    * @param task the task to wait for completion
422    * @throws SQLException if the timeout has expired
423    */
424   public static void waitForTaskCompletion(long timeout,
425       String requestDescription, AbstractTask task) throws SQLException
426   {
427     // Wait for completion (notified by the task)
428 try
429     {
430       // Wait on task
431 if (timeout > 0)
432       {
433         long start = System.currentTimeMillis();
434         task.wait(timeout);
435         long end = System.currentTimeMillis();
436         long remaining = timeout - (end - start);
437         if (remaining <= 0)
438         {
439           if (task.setExpiredTimeout())
440           { // Task will be ignored by all backends
441 String msg = Translate.get("loadbalancer.request.timeout",
442                 new String []{requestDescription,
443                     String.valueOf(task.getSuccess()),
444                     String.valueOf(task.getFailed())});
445
446             logger.warn(msg);
447             throw new SQLException (msg);
448           }
449           // else task execution already started, to late to cancel
450 }
451         // No need to update request timeout since the execution is finished
452 }
453       else
454         task.wait();
455     }
456     catch (InterruptedException e)
457     {
458       if (task.setExpiredTimeout())
459       { // Task will be ignored by all backends
460 String msg = Translate.get("loadbalancer.request.timeout",
461             new String []{requestDescription, String.valueOf(task.getSuccess()),
462                 String.valueOf(task.getFailed())});
463
464         logger.warn(msg);
465         throw new SQLException (msg);
466       }
467       // else task execution already started, to late to cancel
468 }
469   }
470
471   /**
472    * If we are executing in a distributed virtual database, we have to make sure
473    * that we post the query in the queue following the total order. This method
474    * does not remove the request from the total order queue. You have to call
475    * removeHeadFromAndNotifyTotalOrderQueue() to do so.
476    *
477    * @param request the request to wait for (can be any object but usually a
478    * DistributedRequest, Commit or Rollback)
479    * @param errorIfNotFound true if an error message should be logged if the
480    * request is not found in the total order queue
481    * @return true if the element was found and wait has succeeded, false
482    * otherwise
483    * @see #removeObjectFromAndNotifyTotalOrderQueue(Object)
484    */
485   public boolean waitForTotalOrder(Object request, boolean errorIfNotFound)
486   {
487     if (totalOrderQueue != null)
488     {
489       synchronized (totalOrderQueue)
490       {
491         int index = totalOrderQueue.indexOf(request);
492         while (index > 0)
493         {
494           if (logger.isDebugEnabled())
495             logger.debug("Waiting for " + index
496                 + " queries to execute (current is " + totalOrderQueue.get(0)
497                 + ")");
498
499           // All suspended requests can be bypassed
500 boolean foundNonSuspendedRequest = false;
501           for (int i = 0; i < index; i++)
502           {
503             if (!vdb.getRequestManager().getScheduler().isSuspendedRequest(
504                 totalOrderQueue.get(i)))
505             {
506               foundNonSuspendedRequest = true;
507               break;
508             }
509           }
510           if (!foundNonSuspendedRequest)
511           {
512             index = 0;
513             break;
514           }
515
516           try
517           {
518             totalOrderQueue.wait();
519           }
520           catch (InterruptedException ignore)
521           {
522           }
523           index = totalOrderQueue.indexOf(request);
524         }
525         if (index == -1)
526         {
527           if (errorIfNotFound)
528             logger
529                 .error("Request was not found in total order queue, posting out of order ("
530                     + request + ")");
531           return false;
532         }
533         else
534           return true;
535       }
536     }
537     return false;
538   }
539
540   /**
541    * This will block the given request if there are any suspending task in
542    * progress (before the request in the total order queue).
543    *
544    * @param request the request that we are processing
545    */
546   public void waitForSuspendWritesToComplete(AbstractRequest request)
547   {
548     if (totalOrderQueue != null)
549     {
550       synchronized (totalOrderQueue)
551       {
552         boolean hasToWait = true;
553         while (hasToWait)
554         {
555           hasToWait = false;
556           // Checking total order queue to see if there is an
557 // SuspendWritesMessage before this request.
558 // If this is the case, this request will have to wait.
559 for (Iterator iter = totalOrderQueue.iterator(); iter.hasNext();)
560           {
561             Object elem = iter.next();
562             if (elem instanceof SuspendWritesMessage)
563             {
564               // Found a SuspendWritesMessage, so wait...
565 hasToWait = true;
566               break;
567             }
568             else if (elem instanceof AbstractRequest)
569             {
570               // Found the request itself, let'go then...
571 AbstractRequest req = (AbstractRequest) elem;
572               if (req == request)
573                 break;
574             }
575           }
576           if (hasToWait)
577             try
578             {
579               totalOrderQueue.wait();
580             }
581             catch (InterruptedException ignore)
582             {
583             }
584         }
585       }
586     }
587   }
588
589   //
590 // Request Handling
591 //
592
593   /**
594    * Perform a read request. It is up to the implementation to choose to which
595    * backend node(s) this request should be sent.
596    *
597    * @param request an <code>SelectRequest</code>
598    * @param metadataCache MetadataCache (null if none)
599    * @return the corresponding <code>ControllerResultSet</code>
600    * @exception SQLException if an error occurs
601    * @throws AllBackendsFailedException if all backends failed to execute the
602    * request
603    */
604   public abstract ControllerResultSet statementExecuteQuery(
605       SelectRequest request, MetadataCache metadataCache) throws SQLException ,
606       AllBackendsFailedException;
607
608   /**
609    * Perform a write request. This request should usually be broadcasted to all
610    * nodes.
611    *
612    * @param request an <code>AbstractWriteRequest</code>
613    * @return number of rows affected by the request
614    * @throws AllBackendsFailedException if all backends failed to execute the
615    * request
616    * @exception NoMoreBackendException if no backends are left to execute the
617    * request
618    * @exception SQLException if an error occurs
619    */
620   public abstract ExecuteUpdateResult statementExecuteUpdate(
621       AbstractWriteRequest request) throws AllBackendsFailedException,
622       NoMoreBackendException, SQLException ;
623
624   /**
625    * Perform a write request and return a ResultSet containing the auto
626    * generated keys.
627    *
628    * @param request an <code>AbstractWriteRequest</code>
629    * @param metadataCache MetadataCache (null if none)
630    * @return auto generated keys
631    * @throws AllBackendsFailedException if all backends failed to execute the
632    * request
633    * @exception NoMoreBackendException if no backends are left to execute the
634    * request
635    * @exception SQLException if an error occurs
636    */
637   public abstract GeneratedKeysResult statementExecuteUpdateWithKeys(
638       AbstractWriteRequest request, MetadataCache metadataCache)
639       throws AllBackendsFailedException, NoMoreBackendException, SQLException ;
640
641   /**
642    * Call a request that returns multiple results.
643    *
644    * @param request the request to execute
645    * @param metadataCache MetadataCache (null if none)
646    * @return an <code>ExecuteResult</code> object
647    * @throws AllBackendsFailedException if all backends failed to execute the
648    * request
649    * @throws SQLException if an error occurs
650    */
651   public abstract ExecuteResult statementExecute(AbstractRequest request,
652       MetadataCache metadataCache) throws AllBackendsFailedException,
653       SQLException ;
654
655   /**
656    * Call a read-only stored procedure that returns a ResultSet. The stored
657    * procedure will be executed by one node only.
658    *
659    * @param proc the stored procedure call
660    * @param metadataCache MetadataCache (null if none)
661    * @return a <code>ControllerResultSet</code> value
662    * @exception SQLException if an error occurs
663    */
664   public abstract ControllerResultSet readOnlyCallableStatementExecuteQuery(
665       StoredProcedure proc, MetadataCache metadataCache) throws SQLException ;
666
667   /**
668    * Call a read-only stored procedure that returns multiple results. The stored
669    * procedure will be executed by one node only.
670    *
671    * @param proc the stored procedure call
672    * @param metadataCache MetadataCache (null if none)
673    * @return a <code>ExecuteResult</code> object containing all results
674    * @exception SQLException if an error occurs
675    */
676   public abstract ExecuteResult readOnlyCallableStatementExecute(
677       StoredProcedure proc, MetadataCache metadataCache) throws SQLException ;
678
679   /**
680    * Call a stored procedure that returns a ResultSet. This stored procedure can
681    * possibly perform writes and will therefore be executed by all nodes.
682    *
683    * @param proc the stored procedure call
684    * @param metadataCache MetadataCache (null if none)
685    * @return a <code>ControllerResultSet</code> value
686    * @throws AllBackendsFailedException if all backends failed to execute the
687    * request
688    * @exception SQLException if an error occurs
689    */
690   public abstract ControllerResultSet callableStatementExecuteQuery(
691       StoredProcedure proc, MetadataCache metadataCache)
692       throws AllBackendsFailedException, SQLException ;
693
694   /**
695    * Call a stored procedure that performs an update.
696    *
697    * @param proc the stored procedure call
698    * @return number of rows affected
699    * @throws AllBackendsFailedException if all backends failed to execute the
700    * request
701    * @throws SQLException if an error occurs
702    */
703   public abstract ExecuteUpdateResult callableStatementExecuteUpdate(
704       StoredProcedure proc) throws AllBackendsFailedException, SQLException ;
705
706   /**
707    * Call a stored procedure that returns multiple results.
708    *
709    * @param proc the stored procedure call
710    * @param metadataCache MetadataCache (null if none)
711    * @return an <code>ExecuteResult</code> object
712    * @throws AllBackendsFailedException if all backends failed to execute the
713    * request
714    * @throws SQLException if an error occurs
715    */
716   public abstract ExecuteResult callableStatementExecute(StoredProcedure proc,
717       MetadataCache metadataCache) throws AllBackendsFailedException,
718       SQLException ;
719
720   /**
721    * Return a ControllerResultSet containing the PreparedStatement metaData of
722    * the given sql template
723    *
724    * @param request the request containing the sql template
725    * @return an empty ControllerResultSet with the metadata
726    * @throws SQLException if a database error occurs
727    */
728   public abstract ControllerResultSet getPreparedStatementGetMetaData(
729       AbstractRequest request) throws SQLException ;
730
731   /**
732    * Setup a Statement or a PreparedStatement (decoded if parameters of the
733    * request are not null).
734    */
735   private static Statement setupStatementOrPreparedStatement(
736       AbstractRequest request, DatabaseBackend backend,
737       BackendWorkerThread workerThread, Connection c,
738       boolean setupResultSetParameters, boolean needGeneratedKeys)
739       throws SQLException
740   {
741     Statement s; // Can also be used as a PreparedStatement
742 if (request.getPreparedStatementParameters() == null)
743       s = c.createStatement();
744     else
745     {
746       String rewrittenTemplate = backend.rewriteQuery(request
747           .getSqlOrTemplate());
748       if (needGeneratedKeys)
749         s = c.prepareStatement(rewrittenTemplate,
750             Statement.RETURN_GENERATED_KEYS);
751       else
752         s = c.prepareStatement(rewrittenTemplate);
753       PreparedStatementSerialization.setPreparedStatement(request
754           .getPreparedStatementParameters(), (PreparedStatement ) s);
755     }
756
757     // Let the worker thread know which statement we are using in case there is
758 // a need to cancel that statement during its execution
759 if (workerThread != null)
760       workerThread.setCurrentStatement(s);
761
762     DriverCompliance driverCompliance = backend.getDriverCompliance();
763     if (driverCompliance.supportSetQueryTimeout())
764       s.setQueryTimeout(request.getTimeout());
765
766     if (setupResultSetParameters)
767     {
768       if ((request.getCursorName() != null)
769           && (driverCompliance.supportSetCursorName()))
770         s.setCursorName(request.getCursorName());
771       if ((request.getFetchSize() != 0)
772           && driverCompliance.supportSetFetchSize())
773         s.setFetchSize(request.getFetchSize());
774       if ((request.getMaxRows() > 0) && driverCompliance.supportSetMaxRows())
775         s.setMaxRows(request.getMaxRows());
776     }
777     return s;
778   }
779
780   /**
781    * Execute a statement on a backend. If the execution fails, the connection is
782    * checked for validity. If the connection was not valid, the query is
783    * automatically retried on a new connection.<br>
784    *
785    * @param request the request to execute
786    * @param backend the backend on which the request is executed
787    * @param workerThread the backend worker thread executing this query (or null
788    * if none)
789    * @param c connection used to create the statement
790    * @param metadataCache MetadataCache (null if none)
791    * @return the ControllerResultSet
792    * @throws SQLException if an error occurs
793    * @throws BadConnectionException if the connection was bad
794    * @throws UnreachableBackendException if the backend is unreachable
795    */
796   public static final ControllerResultSet executeStatementExecuteQueryOnBackend(
797       SelectRequest request, DatabaseBackend backend,
798       BackendWorkerThread workerThread, Connection c,
799       MetadataCache metadataCache) throws SQLException , BadConnectionException,
800       UnreachableBackendException
801   {
802     ControllerResultSet rs = null;
803     ResultSet backendRS = null;
804     try
805     {
806       backend.addPendingReadRequest(request);
807
808       Statement s = setupStatementOrPreparedStatement(request, backend,
809           workerThread, c, true, false);
810
811       // Execute the query
812 if (request.getPreparedStatementParameters() == null)
813         backendRS = s.executeQuery(backend.rewriteQuery(request
814             .getSqlOrTemplate()));
815       else
816         backendRS = ((PreparedStatement ) s).executeQuery();
817
818       SQLWarning stWarns = null;
819       if (request.getRetrieveSQLWarnings())
820       {
821         stWarns = s.getWarnings();
822       }
823       rs = new ControllerResultSet(request, backendRS, metadataCache, s, false);
824       rs.setStatementWarnings(stWarns);
825     }
826     catch (SQLException e)
827     { // Something bad happened
828 /*
829        * A persistent connection is a direct tie between the client program and
830        * the backend database. If the connection is broken, it cannot be retried
831        * on another connection on the same backend, because the client may be
832        * depending on state that was associated with the broken connection.
833        */
834       if (backend.isValidConnection(c))
835         throw e; // Connection is valid, throw the exception
836 else if (request.isPersistentConnection())
837         throw new UnreachableBackendException("Bad persistent connection", e);
838       else
839         throw new BadConnectionException(e);
840     }
841     finally
842     {
843       // we can close this resultset if fetch size was 0
844 if (backendRS != null && request.getFetchSize() == 0)
845         try
846         {
847           backendRS.close();
848         }
849         catch (SQLException ignore)
850         {
851         }
852       backend.removePendingRequest(request);
853     }
854     return rs;
855   }
856
857   /**
858    * Execute an update prepared statement on a backend. If the execution fails,
859    * the connection is checked for validity. If the connection was not valid,
860    * the query is automatically retried on a new connection.
861    *
862    * @param request the request to execute
863    * @param backend the backend on which the request is executed
864    * @param workerThread the backend worker thread executing this query (or null
865    * if none)
866    * @param c connection used to create the statement
867    * @return int Number of rows effected
868    * @throws SQLException if an error occurs
869    * @throws BadConnectionException if the connection was bad
870    */
871   public static final ExecuteUpdateResult executeStatementExecuteUpdateOnBackend(
872       AbstractWriteRequest request, DatabaseBackend backend,
873       BackendWorkerThread workerThread, Connection c) throws SQLException ,
874       BadConnectionException
875   {
876     Statement s = null;
877     try
878     {
879       backend.addPendingWriteRequest(request);
880
881       s = setupStatementOrPreparedStatement(request, backend, workerThread, c,
882           false, false);
883
884       // Execute the query
885 ExecuteUpdateResult eur;
886       if (request.getPreparedStatementParameters() == null)
887         eur = new ExecuteUpdateResult(s.executeUpdate(backend
888             .rewriteQuery(request.getSqlOrTemplate())));
889       else
890         eur = new ExecuteUpdateResult(((PreparedStatement ) s).executeUpdate());
891       // get warnings, if required
892 if (request.getRetrieveSQLWarnings())
893         eur.setStatementWarnings(s.getWarnings());
894
895       if (request.requiresConnectionPoolFlush())
896         backend.flagAllConnectionsForRenewal();
897
898       return eur;
899     }
900     catch (SQLException e)
901     { // Something bad happened
902 if (backend.isValidConnection(c))
903         throw e; // Connection is valid, throw the exception
904 else
905         throw new BadConnectionException(e);
906     }
907     finally
908     {
909       backend.removePendingRequest(request);
910       try
911       {
912         if (s != null)
913           s.close();
914       }
915       catch (SQLException ignore)
916       {
917       }
918     }
919   }
920
921   /**
922    * Execute an update prepared statement on a backend. If the execution fails,
923    * the connection is checked for validity. If the connection was not valid,
924    * the query is automatically retried on a new connection.
925    *
926    * @param request the request to execute
927    * @param backend the backend on which the request is executed
928    * @param workerThread the backend worker thread executing this query (or null
929    * if none)
930    * @param c connection used to create the statement
931    * @param metadataCache MetadataCache (null if none)
932    * @return ControllerResultSet containing the auto-generated keys
933    * @throws SQLException if an error occurs
934    * @throws BadConnectionException if the connection was bad
935    */
936   public static final GeneratedKeysResult executeStatementExecuteUpdateWithKeysOnBackend(
937       AbstractWriteRequest request, DatabaseBackend backend,
938       BackendWorkerThread workerThread, Connection c,
939       MetadataCache metadataCache) throws SQLException , BadConnectionException
940   {
941     if (!backend.getDriverCompliance().supportGetGeneratedKeys())
942       throw new SQLException ("Backend " + backend.getName()
943           + " does not support RETURN_GENERATED_KEYS");
944
945     Statement s = null;
946     try
947     {
948       backend.addPendingWriteRequest(request);
949
950       s = setupStatementOrPreparedStatement(request, backend, workerThread, c,
951           false, true);
952
953       // Execute the query
954 int updateCount;
955       if (request.getPreparedStatementParameters() == null)
956         updateCount = s.executeUpdate(backend.rewriteQuery(request
957             .getSqlOrTemplate()), Statement.RETURN_GENERATED_KEYS);
958       else
959         updateCount = ((PreparedStatement ) s).executeUpdate();
960       // get warnings, if required
961 SQLWarning stWarns = null;
962       if (request.getRetrieveSQLWarnings())
963       {
964         stWarns = s.getWarnings();
965       }
966       ControllerResultSet rs = new ControllerResultSet(request, s
967           .getGeneratedKeys(), metadataCache, s, false);
968       GeneratedKeysResult gkr = new GeneratedKeysResult(rs, updateCount);
969       gkr.setStatementWarnings(stWarns);
970
971       if (request.requiresConnectionPoolFlush())
972         backend.flagAllConnectionsForRenewal();
973
974       return gkr;
975     }
976     catch (SQLException e)
977     { // Something bad happened
978 if (backend.isValidConnection(c))
979         throw e; // Connection is valid, throw the exception
980 else
981         throw new BadConnectionException(e);
982     }
983     finally
984     {
985       backend.removePendingRequest(request);
986       try
987       {
988         if (s != null)
989           s.close();
990       }
991       catch (SQLException ignore)
992       {
993       }
994     }
995   }
996
997   /**
998    * Execute a request that returns multiple results on the given backend. The
999    * statement is setXXX if the driver has not processed the statement.
1000   *
1001   * @param request the request to execute
1002   * @param backend the backend on which to execute the stored procedure
1003   * @param workerThread the backend worker thread executing this query (or null
1004   * if none)
1005   * @param c the connection on which to execute the stored procedure
1006   * @param metadataCache the matedatacache to build the ControllerResultSet
1007   * @return an <code>ExecuteResult</code> object
1008   * @throws SQLException if an error occurs
1009   * @throws BadConnectionException if the connection was bad
1010   */
1011  public static final ExecuteResult executeStatementExecuteOnBackend(
1012      AbstractRequest request, DatabaseBackend backend,
1013      BackendWorkerThread workerThread, Connection c,
1014      MetadataCache metadataCache) throws SQLException , BadConnectionException
1015  {
1016    Statement s = null;
1017    try
1018    {
1019      backend.addPendingWriteRequest(request);
1020
1021      // Disable fetch size when using execute()
1022 request.setFetchSize(0);
1023
1024      s = setupStatementOrPreparedStatement(request, backend, workerThread, c,
1025          true, false);
1026
1027      // Execute the query
1028 boolean hasResult;
1029      if (request.getPreparedStatementParameters() == null)
1030        hasResult = s.execute(backend.rewriteQuery(request.getSqlOrTemplate()));
1031      else
1032        hasResult = ((PreparedStatement ) s).execute();
1033
1034      int updatedRows = 0;
1035      // Process the result and get all ResultSets or udpate counts
1036 ExecuteResult result = new ExecuteResult();
1037      // get warnings, if required
1038 if (request.getRetrieveSQLWarnings())
1039        result.setStatementWarnings(s.getWarnings());
1040      do
1041      {
1042        if (hasResult)
1043        {
1044          ControllerResultSet crs = new ControllerResultSet(request, s
1045              .getResultSet(), metadataCache, null, true);
1046          result.addResult(crs);
1047        }
1048        else
1049        {
1050          updatedRows = s.getUpdateCount();
1051          result.addResult(updatedRows);
1052        }
1053        hasResult = s.getMoreResults();
1054      }
1055      while (hasResult || (updatedRows != -1));
1056
1057      if (request.requiresConnectionPoolFlush())
1058        backend.flagAllConnectionsForRenewal();
1059
1060      return result;
1061    }
1062    catch (SQLException e)
1063    { // Something bad happened
1064 if (backend.isValidConnection(c))
1065        throw e; // Connection is valid, throw the exception
1066 else
1067        throw new BadConnectionException(e);
1068    }
1069    finally
1070    {
1071      backend.removePendingRequest(request);
1072      try
1073      {
1074        if (s != null)
1075          s.close();
1076      }
1077      catch (SQLException ignore)
1078      {
1079      }
1080    }
1081  }
1082
1083  /**
1084   * Fetch Out and Named parameters if any. The information about the parameters
1085   * is found in the StoredProcedure object and the results are stored in the
1086   * same structures.
1087   * <p>
1088   * After calling this method, the stored procedure object does not contain
1089   * anymore the types of the parameters but their returned values.
1090   *
1091   * @param cs callable statement to fetch from
1092   * @param proc stored procedure object with parameters information
1093   * @throws SQLException if an error occurs during fetching
1094   */
1095  private static void fetchOutAndNamedParameters(CallableStatement cs,
1096      StoredProcedure proc) throws SQLException
1097  {
1098    // First fetch the out parameters
1099 List outParamIndexes = proc.getOutParameterIndexes();
1100    if (outParamIndexes != null)
1101    {
1102      for (Iterator iter = outParamIndexes.iterator(); iter.hasNext();)
1103      {
1104        Object index = iter.next();
1105        if (index instanceof Integer )
1106          proc.setOutParameterValue(index, cs.getObject(((Integer ) index)
1107              .intValue()));
1108        else
1109          // Named OUT parameter
1110 proc.setOutParameterValue(index, cs.getObject((String ) index));
1111      }
1112    }
1113
1114    // Fetch the named parameters
1115 List namedParamNames = proc.getNamedParameterNames();
1116    if (namedParamNames != null)
1117    {
1118      for (Iterator iter = namedParamNames.iterator(); iter.hasNext();)
1119      {
1120        // Overwrite the type with the result (re-use the same map)
1121 String paramName = (String ) iter.next();
1122        proc.setNamedParameterValue(paramName, cs.getObject(paramName));
1123      }
1124    }
1125  }
1126
1127  /**
1128   * Setup a Statement or a PreparedStatement (decoded if a SQL template is
1129   * found in the request).
1130   */
1131  private static CallableStatement setupCallableStatement(StoredProcedure proc,
1132      DatabaseBackend backend, BackendWorkerThread workerThread, Connection c,
1133      boolean setupResultSetParameters) throws SQLException
1134  {
1135    CallableStatement cs; // Can also be used as a PreparedStatement
1136 cs = c.prepareCall(backend.rewriteQuery(proc.getSqlOrTemplate()));
1137    if (proc.getPreparedStatementParameters() != null)
1138      PreparedStatementSerialization.setCallableStatement(backend
1139          .rewriteQuery(proc.getPreparedStatementParameters()), cs, proc);
1140
1141    // Let the worker thread know which statement we are using in case there is
1142 // a need to cancel that statement during its execution
1143 if (workerThread != null)
1144      workerThread.setCurrentStatement(cs);
1145
1146    DriverCompliance driverCompliance = backend.getDriverCompliance();
1147    if (driverCompliance.supportSetQueryTimeout())
1148      cs.setQueryTimeout(proc.getTimeout());
1149
1150    if (setupResultSetParameters)
1151    {
1152      if ((proc.getCursorName() != null)
1153          && (driverCompliance.supportSetCursorName()))
1154        cs.setCursorName(proc.getCursorName());
1155      if ((proc.getFetchSize() != 0) && driverCompliance.supportSetFetchSize())
1156        cs.setFetchSize(proc.getFetchSize());
1157      if ((proc.getMaxRows() > 0) && driverCompliance.supportSetMaxRows())
1158        cs.setMaxRows(proc.getMaxRows());
1159    }
1160    return cs;
1161  }
1162
1163  /**
1164   * Execute a read stored procedure on the given backend. The callable
1165   * statement is setXXX if the driver has not processed the statement.<br>
1166   *
1167   * @param proc the stored procedure to execute
1168   * @param backend the backend on which to execute the stored procedure
1169   * @param workerThread the backend worker thread executing this query (or null
1170   * if none)
1171   * @param c the connection on which to execute the stored procedure
1172   * @param metadataCache the matedatacache to build the ControllerResultSet
1173   * @return the controllerResultSet
1174   * @throws SQLException if an error occurs
1175   * @throws BadConnectionException if the connection was bad
1176   */
1177  public static final ControllerResultSet executeCallableStatementExecuteQueryOnBackend(
1178      StoredProcedure proc, DatabaseBackend backend,
1179      BackendWorkerThread workerThread, Connection c,
1180      MetadataCache metadataCache) throws SQLException , BadConnectionException
1181  {
1182    CallableStatement cs = null;
1183    ResultSet backendRS = null;
1184    try
1185    {
1186      backend.addPendingReadRequest(proc);
1187
1188      cs = setupCallableStatement(proc, backend, workerThread, c, true);
1189
1190      // Execute the query
1191 backendRS = cs.executeQuery();
1192
1193      SQLWarning stWarns = null;
1194      if (proc.getRetrieveSQLWarnings())
1195      {
1196        stWarns = cs.getWarnings();
1197      }
1198      ControllerResultSet rs = new ControllerResultSet(proc, backendRS,
1199          metadataCache, cs, false);
1200      rs.setStatementWarnings(stWarns);
1201      fetchOutAndNamedParameters(cs, proc);
1202
1203      if (proc.requiresConnectionPoolFlush())
1204        backend.flagAllConnectionsForRenewal();
1205
1206      return rs;
1207    }
1208    catch (SQLException e)
1209    { // Something bad happened
1210 if (backend.isValidConnection(c))
1211        throw e; // Connection is valid, throw the exception
1212 else
1213        throw new BadConnectionException(e);
1214    }
1215    finally
1216    {
1217      // we can close this resultset if fetch size was 0
1218 if (backendRS != null && proc.getFetchSize() == 0)
1219      {
1220        try
1221        {
1222          backendRS.close();
1223        }
1224        catch (SQLException ignore)
1225        {
1226        }
1227      }
1228      backend.removePendingRequest(proc);
1229    }
1230  }
1231
1232  /**
1233   * Execute a write stored procedure on the given backend. The callable
1234   * statement is setXXX if the driver has not processed the statement.
1235   *
1236   * @param proc the stored procedure to execute
1237   * @param backend the backend on which to execute the stored procedure
1238   * @param workerThread the backend worker thread executing this query (or null
1239   * if none)
1240   * @param c the connection on which to execute the stored procedure
1241   * @return the number of updated rows
1242   * @throws SQLException if an error occurs
1243   * @throws BadConnectionException if the connection was bad
1244   */
1245  public static final ExecuteUpdateResult executeCallableStatementExecuteUpdateOnBackend(
1246      StoredProcedure proc, DatabaseBackend backend,
1247      BackendWorkerThread workerThread, Connection c) throws SQLException ,
1248      BadConnectionException
1249  {
1250    CallableStatement cs = null;
1251    try
1252    {
1253      backend.addPendingWriteRequest(proc);
1254
1255      cs = setupCallableStatement(proc, backend, workerThread, c, false);
1256
1257      // Execute the query
1258 ExecuteUpdateResult eur = new ExecuteUpdateResult(cs.executeUpdate());
1259      // get warnings, if required
1260 if (proc.getRetrieveSQLWarnings())
1261        eur.setStatementWarnings(cs.getWarnings());
1262
1263      fetchOutAndNamedParameters(cs, proc);
1264
1265      if (proc.requiresConnectionPoolFlush())
1266        backend.flagAllConnectionsForRenewal();
1267
1268      return eur;
1269    }
1270    catch (SQLException e)
1271    { // Something bad happened
1272 if (backend.isValidConnection(c))
1273        throw e; // Connection is valid, throw the exception
1274 else
1275        throw new BadConnectionException(e);
1276    }
1277    finally
1278    {
1279      backend.removePendingRequest(proc);
1280      try
1281      {
1282        if (cs != null)
1283          cs.close();
1284      }
1285      catch (SQLException ignore)
1286      {
1287      }
1288    }
1289  }
1290
1291  /**
1292   * Execute a stored procedure that returns multiple results on the given
1293   * backend. The callable statement is setXXX if the driver has not processed
1294   * the statement.
1295   *
1296   * @param proc the stored procedure to execute
1297   * @param backend the backend on which to execute the stored procedure
1298   * @param workerThread the backend worker thread executing this query (or null
1299   * if none)
1300   * @param c the connection on which to execute the stored procedure
1301   * @param metadataCache the matedatacache to build the ControllerResultSet
1302   * @return an <code>ExecuteResult</code> object
1303   * @throws SQLException if an error occurs
1304   * @throws BadConnectionException if the connection was bad
1305   */
1306  public static final ExecuteResult executeCallableStatementExecuteOnBackend(
1307      StoredProcedure proc, DatabaseBackend backend,
1308      BackendWorkerThread workerThread, Connection c,
1309      MetadataCache metadataCache) throws SQLException , BadConnectionException
1310  {
1311    CallableStatement cs = null;
1312    try
1313    {
1314      backend.addPendingWriteRequest(proc);
1315
1316      // Disable fetch size when using execute()
1317 proc.setFetchSize(0);
1318
1319      cs = setupCallableStatement(proc, backend, workerThread, c, true);
1320
1321      // Execute the query
1322 boolean hasResult = cs.execute();
1323      int updatedRows = 0;
1324      // Process the result and get all ResultSets or udpate counts
1325 ExecuteResult result = new ExecuteResult();
1326      // get warnings, if required
1327 if (proc.getRetrieveSQLWarnings())
1328        result.setStatementWarnings(cs.getWarnings());
1329      do
1330      {
1331        if (hasResult)
1332        {
1333          ControllerResultSet crs = new ControllerResultSet(proc, cs
1334              .getResultSet(), metadataCache, null, true);
1335          result.addResult(crs);
1336        }
1337        else
1338        {
1339          updatedRows = cs.getUpdateCount();
1340          result.addResult(updatedRows);
1341        }
1342        if (updatedRows != -1)
1343          hasResult = cs.getMoreResults();
1344      }
1345      while (hasResult || (updatedRows != -1));
1346
1347      fetchOutAndNamedParameters(cs, proc);
1348
1349      if (proc.requiresConnectionPoolFlush())
1350        backend.flagAllConnectionsForRenewal();
1351
1352      return result;
1353    }
1354    catch (SQLException e)
1355    { // Something bad happened
1356 if (backend.isValidConnection(c))
1357        throw e; // Connection is valid, throw the exception
1358 else
1359        throw new BadConnectionException(e);
1360    }
1361    finally
1362    {
1363      backend.removePendingRequest(proc);
1364      try
1365      {
1366        if (cs != null)
1367          cs.close();
1368      }
1369      catch (SQLException ignore)
1370      {
1371      }
1372    }
1373  }
1374
1375  /**
1376   * Get PreparedStatement metadata before the statement is executed.
1377   *
1378   * @param sqlTemplate the PreparedStatement sql template
1379   * @param backend the backend on which we execute the request
1380   * @param c the connection to create the statement from
1381   * @return an empty ResultSet with the associated metadata
1382   * @throws SQLException if an error occurs
1383   * @throws BadConnectionException if the databsae connection was bad
1384   */
1385  public static final ControllerResultSet preparedStatementGetMetaDataOnBackend(
1386      String sqlTemplate, DatabaseBackend backend, Connection c)
1387      throws SQLException , BadConnectionException
1388  {
1389    try
1390    {
1391      PreparedStatement ps = c.prepareStatement(sqlTemplate);
1392      return new ControllerResultSet(ControllerConstants.CONTROLLER_FACTORY
1393          .getResultSetMetaDataFactory().copyResultSetMetaData(
1394              ps.getMetaData(), null), new ArrayList ());
1395    }
1396    catch (SQLException e)
1397    { // Something bad happened
1398 if (backend.isValidConnection(c))
1399        throw e; // Connection is valid, throw the exception
1400 else
1401        throw new BadConnectionException(e);
1402    }
1403  }
1404
1405  //
1406 // Transaction management
1407 //
1408
1409  /**
1410   * Abort a transaction and all its currently pending or executing queries.
1411   *
1412   * @param tm The transaction marker metadata
1413   * @throws SQLException if an error occurs
1414   */
1415  public abstract void abort(TransactionMetaData tm) throws SQLException ;
1416
1417  /**
1418   * Begin a new transaction.
1419   *
1420   * @param tm The transaction marker metadata
1421   * @throws SQLException if an error occurs
1422   */
1423  public abstract void begin(TransactionMetaData tm) throws SQLException ;
1424
1425  /**
1426   * Commit a transaction.
1427   *
1428   * @param tm The transaction marker metadata
1429   * @throws AllBackendsFailedException if all backends failed to execute the
1430   * request
1431   * @throws SQLException if an error occurs
1432   */
1433  public abstract void commit(TransactionMetaData tm)
1434      throws AllBackendsFailedException, SQLException ;
1435
1436  /**
1437   * Rollback a transaction.
1438   *
1439   * @param tm The transaction marker metadata
1440   * @throws AllBackendsFailedException if all backends failed to execute the
1441   * request
1442   * @throws SQLException if an error occurs
1443   */
1444  public abstract void rollback(TransactionMetaData tm)
1445      throws AllBackendsFailedException, SQLException ;
1446
1447  /**
1448   * Rollback a transaction to a savepoint
1449   *
1450   * @param tm The transaction marker metadata
1451   * @param savepointName The name of the savepoint
1452   * @throws AllBackendsFailedException if all backends failed to execute the
1453   * request
1454   * @throws SQLException if an error occurs
1455   */
1456  public abstract void rollbackToSavepoint(TransactionMetaData tm,
1457      String savepointName) throws AllBackendsFailedException, SQLException ;
1458
1459  /**
1460   * Set a savepoint to a transaction.
1461   *
1462   * @param tm The transaction marker metadata
1463   * @param name The name of the new savepoint
1464   * @throws AllBackendsFailedException if all backends failed to execute the
1465   * request
1466   * @throws SQLException if an error occurs
1467   */
1468  public abstract void setSavepoint(TransactionMetaData tm, String name)
1469      throws AllBackendsFailedException, SQLException ;
1470
1471  /**
1472   * Release a savepoint from a transaction
1473   *
1474   * @param tm The transaction marker metadata
1475   * @param name The name of the savepoint ro release
1476   * @throws AllBackendsFailedException if all backends failed to execute the
1477   * request
1478   * @throws SQLException if an error occurs
1479   */
1480  public abstract void releaseSavepoint(TransactionMetaData tm, String name)
1481      throws AllBackendsFailedException, SQLException ;
1482
1483  /**
1484   * Close a persistent connection.
1485   *
1486   * @param login login requesting the connection closing
1487   * @param persistentConnectionId id of the persistent connection to close
1488   * @throws SQLException if an error occurs
1489   */
1490  public abstract void closePersistentConnection(String login,
1491      long persistentConnectionId) throws SQLException ;
1492
1493  /**
1494   * Open a persistent connection.
1495   *
1496   * @param login login requesting the connection closing
1497   * @param persistentConnectionId id of the persistent connection to open
1498   * @throws SQLException if an error occurs
1499   */
1500  public abstract void openPersistentConnection(String login,
1501      long persistentConnectionId) throws SQLException ;
1502
1503  /**
1504   * Factorized code to start a transaction on a backend and to retrieve a
1505   * connection on this backend
1506   *
1507   * @param backend the backend needed to check valid connection against this
1508   * backend test statement
1509   * @param cm the connection manager to use to retrieve connections
1510   * @param request request that will execute (must carry transaction id and
1511   * transaction isolation level (does nothing if equals to
1512   * Connection.DEFAULT_TRANSACTION_ISOLATION_LEVEL))
1513   * @return a valid connection with a started transaction
1514   * @throws SQLException if the backend is valid but set autocommit cannot be
1515   * set to false
1516   * @throws UnreachableBackendException if the backend is not reachable, ie not
1517   * valid connection can be retrieved
1518   * @see org.continuent.sequoia.driver.Connection#DEFAULT_TRANSACTION_ISOLATION_LEVEL
1519   */
1520  public static final Connection getConnectionAndBeginTransaction(
1521      DatabaseBackend backend, AbstractConnectionManager cm,
1522      AbstractRequest request) throws SQLException , UnreachableBackendException
1523  {
1524    PooledConnection pc = null;
1525    boolean isConnectionValid = false;
1526    Connection c;
1527
1528    do
1529    {
1530      if (request.isPersistentConnection())
1531      { // Retrieve the persistent connection and register it for the
1532 // transaction
1533 pc = cm.retrieveConnectionInAutoCommit(request);
1534        cm.registerConnectionForTransaction(pc, request.getTransactionId());
1535      }
1536      else
1537      { // Get a new connection for the transaction
1538 pc = cm.getConnectionForTransaction(request.getTransactionId());
1539      }
1540
1541      // Sanity check
1542 if (pc == null)
1543      {
1544        throw new UnreachableBackendException(Translate.get(
1545            "loadbalancer.unable.get.connection", new String []{
1546                String.valueOf(request.getTransactionId()), backend.getName()}));
1547      }
1548      c = pc.getConnection();
1549      try
1550      {
1551        if (request.getTransactionIsolation() != org.continuent.sequoia.driver.Connection.DEFAULT_TRANSACTION_ISOLATION_LEVEL)
1552        {
1553          /*
1554           * A user specified transaction isolation will prevail on any other
1555           * settings
1556           */
1557          pc.setTransactionIsolation(request.getTransactionIsolation());
1558        }
1559        else if (defaultTransactionIsolationLevel != org.continuent.sequoia.driver.Connection.DEFAULT_TRANSACTION_ISOLATION_LEVEL)
1560        {
1561          /*
1562           * The defaultTransactionIsolationLevel can be enforced in the
1563           * configuration file to force all transactions to use this level of
1564           * isolation
1565           */
1566          pc.setTransactionIsolation(defaultTransactionIsolationLevel);
1567        }
1568
1569        c.setAutoCommit(false);
1570        isConnectionValid = true;
1571      }
1572      catch (SQLException e)
1573      {
1574        if (backend.isValidConnection(c))
1575          throw e; // Connection is valid, throw the exception
1576 else
1577        {
1578          cm.deleteConnection(request.getTransactionId());
1579          if (request.isPersistentConnection())
1580          {
1581            cm.deletePersistentConnection(request.getPersistentConnectionId());
1582          }
1583        }
1584      }
1585    }
1586    while (!isConnectionValid);
1587    if (pc == null)
1588    {
1589      if (logger.isErrorEnabled())
1590      {
1591        logger.error("Got a null connection [backend=" + backend.getName()
1592            + ", tid=" + request.getTransactionId() + "]");
1593      }
1594    }
1595    return c;
1596  }
1597
1598  //
1599 // Backends management
1600 //
1601
1602  /**
1603   * Enable a backend without further check. The backend is at least read
1604   * enabled but could also be enabled for writes. Ask the corresponding
1605   * connection manager to initialize the connections if needed.
1606   *
1607   * @param db The database backend to enable
1608   * @param writeEnabled True if the backend must be enabled for writes
1609   * @throws SQLException if an error occurs
1610   */
1611  public abstract void enableBackend(DatabaseBackend db, boolean writeEnabled)
1612      throws SQLException ;
1613
1614  /**
1615   * Disable a backend without further check. Ask the corresponding connection
1616   * manager to finalize the connections if needed. This method should not be
1617   * called directly but instead should access the
1618   * <code>RequestManager.disableBackend(...)</code> method.
1619   *
1620   * @param db The database backend to disable
1621   * @param forceDisable true if disable must be forced
1622   * @throws SQLException if an error occurs
1623   */
1624  public abstract void disableBackend(DatabaseBackend db, boolean forceDisable)
1625      throws SQLException ;
1626
1627  /**
1628   * Get the number of currently enabled backends. 0 means that no backend is
1629   * available.
1630   *
1631   * @return number of currently enabled backends
1632   */
1633  public int getNumberOfEnabledBackends()
1634  {
1635    return enabledBackends.size();
1636  }
1637
1638  /**
1639   * Get information about the Request Load Balancer
1640   *
1641   * @return <code>String</code> containing information
1642   */
1643  public abstract String getInformation();
1644
1645  /**
1646   * Get information about the Request Load Balancer in xml
1647   *
1648   * @return <code>String</code> containing information, xml formatted
1649   */
1650  public abstract String getXmlImpl();
1651
1652  /**
1653   * @see org.continuent.sequoia.common.xml.XmlComponent#getXml()
1654   */
1655  public String getXml()
1656  {
1657    StringBuffer info = new StringBuffer ();
1658    info.append("<" + DatabasesXmlTags.ELT_LoadBalancer + " "
1659        + DatabasesXmlTags.ATT_transactionIsolation + "=\"");
1660    switch (defaultTransactionIsolationLevel)
1661    {
1662      case Connection.TRANSACTION_READ_UNCOMMITTED :
1663        info.append(DatabasesXmlTags.VAL_readUncommitted);
1664        break;
1665      case Connection.TRANSACTION_READ_COMMITTED :
1666        info.append(DatabasesXmlTags.VAL_readCommitted);
1667        break;
1668      case Connection.TRANSACTION_REPEATABLE_READ :
1669        info.append(DatabasesXmlTags.VAL_repeatableRead);
1670        break;
1671      case Connection.TRANSACTION_SERIALIZABLE :
1672        info.append(DatabasesXmlTags.VAL_serializable);
1673        break;
1674      default :
1675        info.append(DatabasesXmlTags.VAL_databaseDefault);
1676        break;
1677    }
1678    info.append("\">");
1679    info.append(getXmlImpl());
1680    info.append("</" + DatabasesXmlTags.ELT_LoadBalancer + ">");
1681    return info.toString();
1682  }
1683}
1684

A to Z: JavaDoc & Examples

---

Daily Java News & Articles

---

Open Source Projects

---

Open Source Codes

---

Free Computer Books

---

Remove Frame

---

Popular Tags