Java > Open Source Codes > org > objectweb > cjdbc > controller > requestmanager > distributed > DistributedRequestManager


1 /**
2  * C-JDBC: Clustered JDBC.
3  * Copyright (C) 2002-2005 French National Institute For Research In Computer
4  * Science And Control (INRIA).
5  * Contact: c-jdbc@objectweb.org
6  *
7  * This library is free software; you can redistribute it and/or modify it
8  * under the terms of the GNU Lesser General Public License as published by the
9  * Free Software Foundation; either version 2.1 of the License, or any later
10  * version.
11  *
12  * This library is distributed in the hope that it will be useful, but WITHOUT
13  * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
14  * FITNESS FOR A PARTICULAR PURPOSE. See the GNU Lesser General Public License
15  * for more details.
16  *
17  * You should have received a copy of the GNU Lesser General Public License
18  * along with this library; if not, write to the Free Software Foundation,
19  * Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA.
20  *
21  * Initial developer(s): Emmanuel Cecchet.
22  * Contributor(s): Olivier Fambon, Jean-Bernard van Zuylen.
23  */
24
25 package org.objectweb.cjdbc.controller.requestmanager.distributed;
26
27 import java.sql.SQLException ;
28 import java.util.ArrayList ;
29 import java.util.Vector ;
30
31 import javax.management.NotCompliantMBeanException ;
32
33 import org.objectweb.cjdbc.common.exceptions.NoMoreBackendException;
34 import org.objectweb.cjdbc.common.exceptions.VirtualDatabaseException;
35 import org.objectweb.cjdbc.common.i18n.Translate;
36 import org.objectweb.cjdbc.common.log.Trace;
37 import org.objectweb.cjdbc.common.shared.BackendInfo;
38 import org.objectweb.cjdbc.common.shared.BackendState;
39 import org.objectweb.cjdbc.common.sql.AbstractRequest;
40 import org.objectweb.cjdbc.common.sql.AbstractWriteRequest;
41 import org.objectweb.cjdbc.common.sql.SelectRequest;
42 import org.objectweb.cjdbc.common.sql.StoredProcedure;
43 import org.objectweb.cjdbc.controller.backend.DatabaseBackend;
44 import org.objectweb.cjdbc.controller.cache.result.AbstractResultCache;
45 import org.objectweb.cjdbc.controller.loadbalancer.AbstractLoadBalancer;
46 import org.objectweb.cjdbc.controller.loadbalancer.AllBackendsFailedException;
47 import org.objectweb.cjdbc.controller.recoverylog.RecoveryLog;
48 import org.objectweb.cjdbc.controller.requestmanager.RequestManager;
49 import org.objectweb.cjdbc.controller.requestmanager.TransactionMarkerMetaData;
50 import org.objectweb.cjdbc.controller.scheduler.AbstractScheduler;
51 import org.objectweb.cjdbc.controller.virtualdatabase.ControllerResultSet;
52 import org.objectweb.cjdbc.controller.virtualdatabase.DistributedVirtualDatabase;
53 import org.objectweb.cjdbc.controller.virtualdatabase.VirtualDatabase;
54 import org.objectweb.cjdbc.controller.virtualdatabase.protocol.CJDBCGroupMessage;
55 import org.objectweb.cjdbc.controller.virtualdatabase.protocol.DisableBackend;
56 import org.objectweb.cjdbc.controller.virtualdatabase.protocol.EnableBackend;
57 import org.objectweb.tribe.adapters.MulticastRequestAdapter;
58
59 /**
60  * This class defines a Distributed Request Manager.
61  * <p>
62  * The DRM is composed of a Request Scheduler, an optional Query Cache, and a
63  * Load Balancer and an optional Recovery Log. Unlike a non-dsitributed Request
64  * Manager, this implementation is responsible for synchronizing the different
65  * controllers components (schedulers, ...). Functions that are RAIDb level
66  * dependent are implemented in sub-classes.
67  *
68  * @author <a HREF="mailto:Emmanuel.Cecchet@inria.fr">Emmanuel Cecchet </a>
69  * @author <a HREF="mailto:jbvanzuylen@transwide.com">Jean-Bernard van Zuylen
70  * </a>
71  * @version 1.0
72  */
73 public abstract class DistributedRequestManager extends RequestManager
74 {
75   protected DistributedVirtualDatabase dvdb;
76   /** List of queries that failed on all backends */
77   private Vector failedOnAllBackends;
78   /** Unique controller identifier */
79   private long controllerId;
80   /** List of transactions that have executed on multiple controllers */
81   private ArrayList distributedTransactions;
82
83   // Used to check if a result was received or not
84 protected static final int NO_RESULT = -5;
85
86   /**
87    * Builds a new <code>DistributedRequestManager</code> instance without
88    * cache.
89    *
90    * @param vdb the virtual database this request manager belongs to
91    * @param scheduler the Request Scheduler to use
92    * @param cache a Query Cache implementation
93    * @param loadBalancer the Request Load Balancer to use
94    * @param recoveryLog the Log Recovery to use
95    * @param beginTimeout timeout in seconds for begin
96    * @param commitTimeout timeout in seconds for commit
97    * @param rollbackTimeout timeout in seconds for rollback
98    * @throws SQLException if an error occurs
99    * @throws NotCompliantMBeanException if this class is not a compliant JMX
100    * MBean
101    */
102   public DistributedRequestManager(DistributedVirtualDatabase vdb,
103       AbstractScheduler scheduler, AbstractResultCache cache,
104       AbstractLoadBalancer loadBalancer, RecoveryLog recoveryLog,
105       long beginTimeout, long commitTimeout, long rollbackTimeout)
106       throws SQLException , NotCompliantMBeanException
107   {
108     super(vdb, scheduler, cache, loadBalancer, recoveryLog, beginTimeout,
109         commitTimeout, rollbackTimeout);
110     dvdb = vdb;
111     failedOnAllBackends = new Vector ();
112     distributedTransactions = new ArrayList ();
113   }
114
115   //
116 // Controller identifier related functions
117 //
118
119   /**
120    * Effective controllerIds are on the upper 16 bits of a long (64 bits).
121    * Distributed transaction ids (longs) are layed out as [ControllerId(16bits) |
122    * LocalTransactionId(64bits)]. <br/>This constant used in
123    * DistributedVirtualDatabase.
124    */
125   public static final long CONTROLLER_ID_BIT_MASK = 0xffff000000000000L;
126   /**
127    * TRANSACTION_ID_BIT_MASK is used to get the transaction id local to the
128    * originating controller
129    */
130   public static final long TRANSACTION_ID_BIT_MASK = ~CONTROLLER_ID_BIT_MASK;
131
132   /**
133    * @see #CONTROLLER_ID_BIT_MASK
134    */
135   public static final int CONTROLLER_ID_SHIFT_BITS = 48;
136
137   /**
138    * @see #CONTROLLER_ID_BIT_MASK
139    */
140   public static final long CONTROLLER_ID_BITS = 0x000000000000ffffL;
141
142   /**
143    * Returns the unique controller identifier.
144    *
145    * @return Returns the controllerId.
146    */
147   public long getControllerId()
148   {
149     return controllerId;
150   }
151
152   /**
153    * Sets the controller identifier value (this id must be unique). Parameter id
154    * must hold on 16 bits (&lt; 0xffff), otherwise an exception is thrown.
155    * Effective this.controllerId is <strong>not </strong> set to passed
156    * parameter id, but to id &lt;&lt; ControllerIdShiftBits. The reason for all
157    * this is that controllerIds are to be carried into ditributed transactions
158    * ids, in the upper 16 bits.
159    *
160    * @param id The controllerId to set.
161    */
162   public void setControllerId(long id)
163   {
164     if ((id & ~CONTROLLER_ID_BITS) != 0)
165     {
166       String msg = "Out of range controller id (" + id + ")";
167       logger.error(msg);
168       throw new RuntimeException (msg);
169     }
170     this.controllerId = (id << CONTROLLER_ID_SHIFT_BITS)
171         & CONTROLLER_ID_BIT_MASK;
172     if (logger.isDebugEnabled())
173       logger.debug("Setting controller identifier to " + id
174           + " (shifted value is " + controllerId + ")");
175   }
176
177   /**
178    * Get the trace logger of this DistributedRequestManager
179    *
180    * @return a <code>Trace</code> object
181    */
182   public Trace getLogger()
183   {
184     return logger;
185   }
186
187   /**
188    * Returns the vdb value.
189    *
190    * @return Returns the vdb.
191    */
192   public VirtualDatabase getVirtualDatabase()
193   {
194     return dvdb;
195   }
196
197   /**
198    * @see org.objectweb.cjdbc.controller.requestmanager.RequestManager#setScheduler(org.objectweb.cjdbc.controller.scheduler.AbstractScheduler)
199    */
200   public void setScheduler(AbstractScheduler scheduler)
201   {
202     super.setScheduler(scheduler);
203     // Note: don't try to use this.dvdb here: setScheduler is called by the
204 // c'tor, and dvdb is not set at this time.
205 if (vdb.getTotalOrderQueue() == null)
206       throw new RuntimeException (
207           "New scheduler does not support total ordering and is not compatible with distributed virtual databases.");
208   }
209
210   //
211 // Database Backends management
212 //
213
214   /**
215    * Enable a backend that has been previously added to this virtual database
216    * and that is in the disabled state. We check we the other controllers if
217    * this backend must be enabled in read-only or read-write. The current policy
218    * is that the first one to enable this backend will have read-write access to
219    * it and others will be in read-only.
220    *
221    * @param db The database backend to enable
222    * @throws SQLException if an error occurs
223    */
224   public void enableBackend(DatabaseBackend db) throws SQLException
225   {
226     int size = dvdb.getAllMemberButUs().size();
227     if (size > 0)
228     {
229       logger.debug(Translate
230           .get("virtualdatabase.distributed.enable.backend.check"));
231
232       try
233       {
234         // Notify other controllers that we enable this backend.
235 // No answer is expected.
236 dvdb.getMulticastRequestAdapter().multicastMessage(
237             dvdb.getAllMemberButUs(), new EnableBackend(new BackendInfo(db)),
238             MulticastRequestAdapter.WAIT_NONE,
239             CJDBCGroupMessage.defaultCastTimeOut);
240       }
241       catch (Exception e)
242       {
243         String msg = "Error while enabling backend " + db.getName();
244         logger.error(msg, e);
245         throw new SQLException (msg + "(" + e + ")");
246       }
247     }
248
249     super.enableBackend(db);
250   }
251
252   /**
253    * @see org.objectweb.cjdbc.controller.requestmanager.RequestManager#disableBackend(org.objectweb.cjdbc.controller.backend.DatabaseBackend)
254    */
255   public void disableBackend(DatabaseBackend db) throws SQLException
256   {
257     int size = dvdb.getAllMemberButUs().size();
258     if (size > 0)
259     {
260       logger.debug(Translate.get("virtualdatabase.distributed.disable.backend",
261           db.getName()));
262
263       try
264       {
265         // Notify other controllers that we disable this backend.
266 // No answer is expected.
267 dvdb.getMulticastRequestAdapter().multicastMessage(
268             dvdb.getAllMemberButUs(), new DisableBackend(new BackendInfo(db)),
269             MulticastRequestAdapter.WAIT_NONE,
270             CJDBCGroupMessage.defaultCastTimeOut);
271       }
272       catch (Exception e)
273       {
274         String msg = "Error while disabling backend " + db.getName();
275         logger.error(msg, e);
276         throw new SQLException (msg + "(" + e + ")");
277       }
278     }
279
280     super.disableBackend(db);
281   }
282
283   /**
284    * The backend must belong to this virtual database and be in the enabled
285    * state.
286    * <p>
287    * The backend is disabled once all the pending write queries are executed. A
288    * checkpoint is inserted in the recovery log.
289    *
290    * @param db The database backend to enable
291    * @param checkpointName The checkpoint name to restart from
292    * @throws SQLException if an error occurs
293    */
294   public void disableBackendForCheckpoint(DatabaseBackend db,
295       String checkpointName) throws SQLException
296   {
297     // Sanity checks
298 if (recoveryLog == null)
299     {
300       String msg = Translate.get("recovery.store.checkpoint.failed.cause.null",
301           checkpointName);
302       logger.error(msg);
303       throw new SQLException (msg);
304     }
305
306     // Set dvdb-wide (distributed) checkpoint before we disable the backend.
307 try
308     {
309       dvdb.setGroupCheckpoint(checkpointName, dvdb.getAllMembers());
310     }
311     catch (VirtualDatabaseException e)
312     {
313       String msg = "set group checkpoint failed";
314       logger.error(msg, e);
315       throw new SQLException (msg);
316     }
317
318     // Signal the backend should not begin any new transaction
319 db.setState(BackendState.DISABLING);
320     logger.info(Translate.get("backend.state.disabling", db.getName()));
321
322     // Wait for all current transactions on the backend to finish
323 db.waitForAllTransactionsToComplete();
324
325     // Now we can safely disable the backend
326 db.setLastKnownCheckpoint(checkpointName);
327     loadBalancer.disableBackend(db);
328     logger.info(Translate.get("backend.state.disabled", db.getName()));
329
330   }
331
332   /**
333    * Add a request that failed on all backends.
334    *
335    * @param request the request that failed
336    * @see #completeFailedOnAllBackends(AbstractRequest, boolean)
337    */
338   public void addFailedOnAllBackends(AbstractRequest request)
339   {
340     failedOnAllBackends.add(request);
341   }
342
343   /**
344    * Notify completion of a request that failed on all backends. If completion
345    * was successful, all local backends are disabled.
346    *
347    * @param request request that completed
348    * @param success true if completion is successful
349    * @see #addFailedOnAllBackends(AbstractRequest)
350    */
351   public void completeFailedOnAllBackends(AbstractRequest request,
352       boolean success)
353   {
354     if (!failedOnAllBackends.remove(request))
355     {
356       logger.warn("Unable to find request "
357           + request.getSQLShortForm(dvdb.getSQLShortFormLength())
358           + " in list of requests that failed on all backends.");
359       return;
360     }
361     if (success)
362     { // We have to invalidate all backends
363 logger
364           .error("Request "
365               + request.getSQLShortForm(dvdb.getSQLShortFormLength())
366               + " failed on all local backends but succeeded on other controllers. Disabling all local backends.");
367       try
368       {
369         dvdb.disableAllBackends();
370       }
371       catch (VirtualDatabaseException e)
372       {
373         logger.error("An error occured while disabling all backends", e);
374       }
375     }
376     else
377       // Notify scheduler now, the notification was postponed in
378 // ExecWriteRequest or ExecWriteRequestWithKeys
379 scheduler.notifyWriteCompleted((AbstractWriteRequest) request);
380   }
381
382   /**
383    * Remove a request that was logged because no backend was available locally
384    * to execute it but that finally ended up in failing at all other
385    * controllers.
386    *
387    * @param request request that was logged but failed at all controllers
388    * @param recoveryLogId request identifier in the recovery log on controller
389    * where the request should be unlogged
390    */
391   public void removeFailedRequestFromRecoveryLog(AbstractWriteRequest request,
392       long recoveryLogId)
393   {
394     if (logger.isDebugEnabled())
395       logger.debug("Request "
396           + request.getSQLShortForm(dvdb.getSQLShortFormLength())
397           + " failed at all controllers, removing it from recovery log.");
398
399     // Now that the request object has been deserialized here, we have our own
400 // copy and we can safely override the id. This is not possible on the
401 // sender side in DistributedVirtualDatabase.
402 request.setId(recoveryLogId);
403     recoveryLog.unlogRequest(request);
404   }
405
406   /**
407    * Remove a stored procedure call that was logged because no backend was
408    * available locally to execute it but that finally ended up in failing at all
409    * other controllers.
410    *
411    * @param proc stored procedure that was logged but failed at all controllers
412    */
413   public void removeFailedStoredProcedureFromRecoveryLog(StoredProcedure proc)
414   {
415     if (logger.isDebugEnabled())
416       logger.debug("Request "
417           + proc.getSQLShortForm(dvdb.getSQLShortFormLength())
418           + " failed at all controllers, removing it from recovery log.");
419
420     recoveryLog.unlogRequest(proc);
421   }
422
423   /**
424    * Remove a commit call that was logged because no backend was available
425    * locally to execute it but that finally ended up in failing at all other
426    * controllers.
427    *
428    * @param tm the identifier of the transaction that failed to commit
429    */
430   public void removeFailedCommitFromRecoveryLog(TransactionMarkerMetaData tm)
431   {
432     if (logger.isDebugEnabled())
433       logger
434           .debug("Transaction "
435               + tm.getTransactionId()
436               + " commit failed at all controllers, removing it from recovery log.");
437
438     recoveryLog.unlogCommit(tm);
439   }
440
441   /**
442    * Remove a rollback call that was logged because no backend was available
443    * locally to execute it but that finally ended up in failing at all other
444    * controllers.
445    *
446    * @param tm the identifier of the transaction that failed to rollback
447    */
448   public void removeFailedRollbackFromRecoveryLog(TransactionMarkerMetaData tm)
449   {
450     if (logger.isDebugEnabled())
451       logger
452           .debug("Transaction "
453               + tm.getTransactionId()
454               + " rollback failed at all controllers, removing it from recovery log.");
455
456     recoveryLog.unlogRollback(tm);
457   }
458
459   //
460 // Transaction management
461 //
462
463   /**
464    * @see org.objectweb.cjdbc.controller.requestmanager.RequestManager#begin(java.lang.String)
465    */
466   public long begin(String login) throws SQLException
467   {
468     try
469     {
470       TransactionMarkerMetaData tm = new TransactionMarkerMetaData(0,
471           beginTimeout, login);
472
473       // Wait for the scheduler to give us the authorization to execute
474 long tid = scheduler.begin(tm);
475       // 2 first bytes are used for controller id
476 // 6 right-most bytes are used for transaction id
477 tid = tid & TRANSACTION_ID_BIT_MASK;
478       tid = tid | controllerId;
479       tm.setTransactionId(tid);
480
481       if (logger.isDebugEnabled())
482         logger.debug(Translate.get("transaction.begin", String.valueOf(tid)));
483
484       try
485       {
486         // Send to load balancer
487 loadBalancer.begin(tm);
488       }
489       catch (SQLException e)
490       {
491         throw e;
492       }
493       finally
494       {
495         // Notify scheduler for completion in any case
496 scheduler.beginCompleted(tid);
497       }
498
499       tidLoginTable.put(new Long (tid), tm);
500       return tid;
501     }
502     catch (RuntimeException e)
503     {
504       logger.fatal(Translate.get(
505           "fatal.runtime.exception.requestmanager.begin", e));
506       throw new SQLException (e.getMessage());
507     }
508   }
509
510   /**
511    * Check if the transaction corresponding to the given query has been started
512    * remotely and start the transaction locally in a lazy manner if needed.
513    *
514    * @param request query to execute
515    * @throws SQLException if an error occurs
516    */
517   public void lazyTransactionStart(AbstractRequest request) throws SQLException
518   {
519     // Check if this is a remotely started transaction that we need to lazyly
520 // start locally
521 if (!request.isAutoCommit())
522     {
523       long tid = request.getTransactionId();
524       if ((tid & CONTROLLER_ID_BIT_MASK) != controllerId)
525       { // Remote transaction, check that it is started
526 if (!tidLoginTable.containsKey(new Long (tid)))
527         { // Begin this transaction
528 try
529           {
530             TransactionMarkerMetaData tm = new TransactionMarkerMetaData(0,
531                 beginTimeout, request.getLogin());
532             tm.setTransactionId(tid);
533
534             if (logger.isDebugEnabled())
535               logger.debug(Translate.get("transaction.begin.lazy", String
536                   .valueOf(tid)));
537
538             try
539             {
540               scheduler.begin(tm);
541
542               // Send to load balancer
543 loadBalancer.begin(tm);
544
545               // We need to update the tid table first so that
546 // logLazyTransactionBegin can retrieve the metadata
547 tidLoginTable.put(new Long (tid), tm);
548               if (recoveryLog != null)
549                 logLazyTransactionBegin(tid);
550             }
551             catch (SQLException e)
552             {
553               if (recoveryLog != null)
554                 // In case logLazyTransactionBegin failed
555 tidLoginTable.remove(new Long (tid));
556               throw e;
557             }
558             finally
559             {
560               // Notify scheduler for completion in any case
561 scheduler.beginCompleted(tid);
562             }
563           }
564           catch (RuntimeException e)
565           {
566             logger.fatal(Translate.get(
567                 "fatal.runtime.exception.requestmanager.begin", e));
568             throw new SQLException (e.getMessage());
569           }
570         }
571       }
572     }
573   }
574
575   /**
576    * @see org.objectweb.cjdbc.controller.requestmanager.RequestManager#commit(long,
577    * boolean)
578    */
579   public void commit(long transactionId, boolean logCommit) throws SQLException
580   {
581     Long lTid = new Long (transactionId);
582     boolean isAWriteTransaction;
583     synchronized (distributedTransactions)
584     {
585       isAWriteTransaction = distributedTransactions.remove(lTid);
586     }
587     if (isAWriteTransaction)
588     {
589       TransactionMarkerMetaData tm = getTransactionMarker(lTid);
590       distributedCommit(tm.getLogin(), transactionId);
591     }
592     else
593       // read-only transaction, it is local
594 super.commit(transactionId, logCommit);
595   }
596
597   /**
598    * @see org.objectweb.cjdbc.controller.requestmanager.RequestManager#rollback(long,
599    * boolean)
600    */
601   public void rollback(long transactionId, boolean logRollback)
602       throws SQLException
603   {
604     Long lTid = new Long (transactionId);
605     boolean isAWriteTransaction;
606     synchronized (distributedTransactions)
607     {
608       isAWriteTransaction = distributedTransactions.remove(lTid);
609     }
610     if (isAWriteTransaction)
611     {
612       TransactionMarkerMetaData tm = getTransactionMarker(lTid);
613       distributedRollback(tm.getLogin(), transactionId);
614     }
615     else
616       // read-only transaction, it is local
617 super.rollback(transactionId, logRollback);
618   }
619
620   /**
621    * @see org.objectweb.cjdbc.controller.requestmanager.RequestManager#rollback(long,
622    * String)
623    */
624   public void rollback(long transactionId, String savepointName)
625       throws SQLException
626   {
627     Long lTid = new Long (transactionId);
628     boolean isAWriteTransaction;
629     synchronized (distributedTransactions)
630     {
631       isAWriteTransaction = distributedTransactions.contains(lTid);
632     }
633     if (isAWriteTransaction)
634       distributedRollback(transactionId, savepointName);
635     else
636       // read-only transaction, it is local
637 super.rollback(transactionId, savepointName);
638   }
639
640   /**
641    * @see org.objectweb.cjdbc.controller.requestmanager.RequestManager#setSavepoint(long)
642    */
643   public int setSavepoint(long transactionId) throws SQLException
644   {
645     Long lTid = new Long (transactionId);
646     boolean isAWriteTransaction;
647     synchronized (distributedTransactions)
648     {
649       isAWriteTransaction = distributedTransactions.contains(lTid);
650     }
651     if (isAWriteTransaction)
652     {
653       int savepointId = scheduler.incrementSavepointId();
654       distributedSetSavepoint(transactionId, String.valueOf(savepointId));
655       return savepointId;
656     }
657     else
658       // read-only transaction, it is local
659 return super.setSavepoint(transactionId);
660   }
661
662   /**
663    * @see org.objectweb.cjdbc.controller.requestmanager.RequestManager#setSavepoint(long,
664    * String)
665    */
666   public void setSavepoint(long transactionId, String name) throws SQLException
667   {
668     Long lTid = new Long (transactionId);
669     boolean isAWriteTransaction;
670     synchronized (distributedTransactions)
671     {
672       isAWriteTransaction = distributedTransactions.contains(lTid);
673     }
674     if (isAWriteTransaction)
675       distributedSetSavepoint(transactionId, name);
676     else
677       // read-only transaction, it is local
678 super.setSavepoint(transactionId, name);
679   }
680
681   /**
682    * @see org.objectweb.cjdbc.controller.requestmanager.RequestManager#releaseSavepoint(long,
683    * String)
684    */
685   public void releaseSavepoint(long transactionId, String name)
686       throws SQLException
687   {
688     Long lTid = new Long (transactionId);
689     boolean isAWriteTransaction;
690     synchronized (distributedTransactions)
691     {
692       isAWriteTransaction = distributedTransactions.contains(lTid);
693     }
694     if (isAWriteTransaction)
695       distributedReleaseSavepoint(transactionId, name);
696     else
697       // read-only transaction, it is local
698 super.releaseSavepoint(transactionId, name);
699   }
700
701   /**
702    * @see org.objectweb.cjdbc.controller.requestmanager.RequestManager#scheduleExecWriteRequest(org.objectweb.cjdbc.common.sql.AbstractWriteRequest)
703    */
704   public void scheduleExecWriteRequest(AbstractWriteRequest request)
705       throws SQLException
706   {
707     lazyTransactionStart(request);
708     super.scheduleExecWriteRequest(request);
709   }
710
711   /**
712    * @see org.objectweb.cjdbc.controller.requestmanager.RequestManager#execReadRequest(org.objectweb.cjdbc.common.sql.SelectRequest)
713    */
714   public ControllerResultSet execReadRequest(SelectRequest request)
715       throws SQLException
716   {
717     try
718     {
719       return execLocalReadRequest(request);
720     }
721     catch (NoMoreBackendException ignored) // other SQLException thrown
722 {
723       // We have no local backend available to execute the request: try on
724 // other controllers.
725 // Send our identity to avoid a loop in case of boucing read requests
726 // (no backends at all in the system).
727 return execRemoteReadRequest(request);
728     }
729   }
730
731   /**
732    * Execute a read request on some remote controller - one in the group. Used
733    * when the local controller has no backend available to execute the request.
734    *
735    * @param request the request to execute
736    * @return the query ResultSet
737    * @throws SQLException in case of bad request
738    */
739   public abstract ControllerResultSet execRemoteReadRequest(
740       SelectRequest request) throws SQLException ;
741
742   /**
743    * @see org.objectweb.cjdbc.controller.requestmanager.RequestManager#execWriteRequest(org.objectweb.cjdbc.common.sql.AbstractWriteRequest)
744    */
745   public int execWriteRequest(AbstractWriteRequest request) throws SQLException
746   {
747     if (!request.isAutoCommit())
748     { // Add this transaction to the list of write transactions
749 Long lTid = new Long (request.getTransactionId());
750       synchronized (distributedTransactions)
751       {
752         if (!distributedTransactions.contains(lTid))
753           distributedTransactions.add(lTid);
754       }
755     }
756     return execDistributedWriteRequest(request);
757   }
758
759   /**
760    * @see org.objectweb.cjdbc.controller.requestmanager.RequestManager#execWriteRequestWithKeys(org.objectweb.cjdbc.common.sql.AbstractWriteRequest)
761    */
762   public ControllerResultSet execWriteRequestWithKeys(
763       AbstractWriteRequest request) throws SQLException
764   {
765     if (!request.isAutoCommit())
766     { // Add this transaction to the list of write transactions
767 Long lTid = new Long (request.getTransactionId());
768       synchronized (distributedTransactions)
769       {
770         if (!distributedTransactions.contains(lTid))
771           distributedTransactions.add(lTid);
772       }
773     }
774     return execDistributedWriteRequestWithKeys(request);
775   }
776
777   /**
778    * @see org.objectweb.cjdbc.controller.requestmanager.RequestManager#execReadStoredProcedure(StoredProcedure)
779    */
780   public ControllerResultSet execReadStoredProcedure(StoredProcedure proc)
781       throws SQLException
782   {
783     // If connection is read-only, we don't broadcast
784 if (proc.isReadOnly())
785     {
786       try
787       {
788         return execReadStoredProcedureLocally(proc);
789       }
790       catch (AllBackendsFailedException ignore)
791       {
792         // This failed locally, try it remotely
793 }
794     }
795
796     if (!proc.isAutoCommit())
797     { // Add this transaction to the list of write transactions
798 Long lTid = new Long (proc.getTransactionId());
799       synchronized (distributedTransactions)
800       {
801         if (!distributedTransactions.contains(lTid))
802           distributedTransactions.add(lTid);
803       }
804     }
805     return execDistributedReadStoredProcedure(proc);
806   }
807
808   /**
809    * @see org.objectweb.cjdbc.controller.requestmanager.RequestManager#execWriteStoredProcedure(org.objectweb.cjdbc.common.sql.StoredProcedure)
810    */
811   public int execWriteStoredProcedure(StoredProcedure proc) throws SQLException
812   {
813     if (!proc.isAutoCommit())
814     { // Add this transaction to the list of write transactions
815 Long lTid = new Long (proc.getTransactionId());
816       synchronized (distributedTransactions)
817       {
818         if (!distributedTransactions.contains(lTid))
819           distributedTransactions.add(lTid);
820       }
821     }
822     return execDistributedWriteStoredProcedure(proc);
823   }
824
825   //
826 // RAIDb level specific methods
827 //
828
829   /**
830    * Distributed implementation of a commit
831    *
832    * @param login login that commit the transaction
833    * @param transactionId id of the commiting transaction
834    * @throws SQLException if an error occurs
835    */
836   public abstract void distributedCommit(String login, long transactionId)
837       throws SQLException ;
838
839   /**
840    * Distributed implementation of a rollback
841    *
842    * @param login login that rollback the transaction
843    * @param transactionId id of the rollbacking transaction
844    * @throws SQLException if an error occurs
845    */
846   public abstract void distributedRollback(String login, long transactionId)
847       throws SQLException ;
848
849   /**
850    * Distributed implementation of a rollback to a savepoint
851    *
852    * @param transactionId id of the transaction
853    * @param savepointName name of the savepoint
854    * @throws SQLException if an error occurs
855    */
856   public abstract void distributedRollback(long transactionId,
857       String savepointName) throws SQLException ;
858
859   /**
860    * Distributed implementation of setting a savepoint to a transaction
861    *
862    * @param transactionId id of the transaction
863    * @param name name of the savepoint to set
864    * @throws SQLException if an error occurs
865    */
866   public abstract void distributedSetSavepoint(long transactionId, String name)
867       throws SQLException ;
868
869   /**
870    * Distributed implementation of releasing a savepoint from a transaction
871    *
872    * @param transactionId id of the transaction
873    * @param name name of the savepoint to release
874    * @throws SQLException if an error occurs
875    */
876   public abstract void distributedReleaseSavepoint(long transactionId,
877       String name) throws SQLException ;
878
879   /**
880    * Distributed implementation of a write request execution.
881    *
882    * @param request request to execute
883    * @return number of modified rows
884    * @throws SQLException if an error occurs
885    */
886   public abstract int execDistributedWriteRequest(AbstractWriteRequest request)
887       throws SQLException ;
888
889   /**
890    * Distributed implementation of a write request execution that returns
891    * auto-generated keys.
892    *
893    * @param request request to execute
894    * @return ResultSet containing the auto-generated keys.
895    * @throws SQLException if an error occurs
896    */
897   public abstract ControllerResultSet execDistributedWriteRequestWithKeys(
898       AbstractWriteRequest request) throws SQLException ;
899
900   /**
901    * Distributed implementation of a read stored procedure execution.
902    *
903    * @param proc stored procedure to execute
904    * @return ResultSet corresponding to this stored procedure execution
905    * @throws SQLException if an error occurs
906    */
907   public abstract ControllerResultSet execDistributedReadStoredProcedure(
908       StoredProcedure proc) throws SQLException ;
909
910   /**
911    * Distributed implementation of a write stored procedure execution.
912    *
913    * @param proc stored procedure to execute
914    * @return number of modified rows
915    * @throws SQLException if an error occurs
916    */
917   public abstract int execDistributedWriteStoredProcedure(StoredProcedure proc)
918       throws SQLException ;
919
920   /**
921    * Once the request has been dispatched, it can be executed using the code
922    * from <code>RequestManager</code>
923    *
924    * @param proc stored procedure to execute
925    * @return ResultSet corresponding to this stored procedure execution
926    * @throws AllBackendsFailedException if all backends failed to execute the
927    * stored procedure
928    * @throws SQLException if an error occurs
929    */
930   public ControllerResultSet execReadStoredProcedureLocally(StoredProcedure proc)
931       throws AllBackendsFailedException, SQLException
932   {
933     return super.execReadStoredProcedure(proc);
934   }
935
936   /**
937    * Once the request has been dispatched, it can be executed using the code
938    * from <code>RequestManager</code>
939    *
940    * @param proc stored procedure to execute
941    * @return number of modified rows
942    * @throws AllBackendsFailedException if all backends failed to execute the
943    * stored procedure
944    * @throws SQLException if an error occurs
945    */
946   public int execDistributedWriteStoredProcedureLocally(StoredProcedure proc)
947       throws AllBackendsFailedException, SQLException
948   {
949     return super.execWriteStoredProcedure(proc);
950   }
951
952   /**
953    * Performs a local read operation, as opposed to execReadRequest() which
954    * attempts to use distributed reads when there is NoMoreBackendException.
955    *
956    * @param request the read request to perform
957    * @return a ControllerResultSet
958    * @throws NoMoreBackendException when no more local backends are available to
959    * execute the request
960    * @throws SQLException in case of error
961    */
962   public ControllerResultSet execLocalReadRequest(SelectRequest request)
963       throws NoMoreBackendException, SQLException
964   {
965     return super.execReadRequest(request);
966   }
967
968 }

A to Z: JavaDoc & Examples

---

Daily Java News & Articles

---

Open Source Projects

---

Open Source Codes

---

Free Computer Books

---

Remove Frame

---

Popular Tags