RAIDb1OptimisticTransactionLevelScheduler


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): Jean-Bernard van Zuylen.
23   */
24  
25  package org.objectweb.cjdbc.controller.scheduler.raidb1;
26  
27  import java.sql.SQLException  ;
28  import java.util.ArrayList  ;
29  
30  import org.objectweb.cjdbc.common.exceptions.RollbackException;
31  import org.objectweb.cjdbc.common.sql.AbstractWriteRequest;
32  import org.objectweb.cjdbc.common.sql.ParsingGranularities;
33  import org.objectweb.cjdbc.common.sql.SelectRequest;
34  import org.objectweb.cjdbc.common.sql.StoredProcedure;
35  import org.objectweb.cjdbc.common.sql.schema.DatabaseSchema;
36  import org.objectweb.cjdbc.common.sql.schema.DatabaseTable;
37  import org.objectweb.cjdbc.common.xml.DatabasesXmlTags;
38  import org.objectweb.cjdbc.controller.requestmanager.RAIDbLevels;
39  import org.objectweb.cjdbc.controller.scheduler.AbstractScheduler;
40  import org.objectweb.cjdbc.controller.scheduler.schema.SchedulerDatabaseSchema;
41  import org.objectweb.cjdbc.controller.scheduler.schema.SchedulerDatabaseTable;
42  import org.objectweb.cjdbc.controller.scheduler.schema.TransactionExclusiveLock;
43  
44  /**
45   * This scheduler provides transaction level scheduling for RAIDb-1 controllers.
46   * Each write takes a lock on the table it affects. All following writes are
47   * blocked until the transaction of the first write completes. This scheduler
48   * automatically detects simple deadlocks and rollbacks the transaction inducing
49   * the deadlock. Note that transitive deadlocks (involving more than 2 tables
50   * are not detected).
51   * 
52   * @author <a HREF="mailto:Emmanuel.Cecchet@inria.fr">Emmanuel Cecchet </a>
53   * @author <a HREF="mailto:jbvanzuylen@transwide.com">Jean-Bernard van Zuylen
54   *         </a>
55   * @version 1.0
56   */
57  public class RAIDb1OptimisticTransactionLevelScheduler
58      extends AbstractScheduler
59  {
60  
61    //
62    // How the code is organized ?
63    //
64    // 1. Member variables
65    // 2. Constructor
66    // 3. Request handling
67    // 4. Transaction management
68    // 5. Debug/Monitoring
69    //
70  
71    private long                    requestId;
72    private SchedulerDatabaseSchema schedulerDatabaseSchema = null;
73  
74    //
75    // Constructor
76    //
77  
78    /**
79     * Creates a new Optimistic Transaction Level Scheduler
80     */
81    public RAIDb1OptimisticTransactionLevelScheduler()
82    {
83      super(RAIDbLevels.RAIDb1, ParsingGranularities.TABLE);
84      requestId = 0;
85    }
86  
87    //
88    // Request Handling
89    //
90  
91    /**
92     * Sets the <code>DatabaseSchema</code> of the current virtual database.
93     * This is only needed by some schedulers that will have to define their own
94     * scheduler schema
95     * 
96     * @param dbs a <code>DatabaseSchema</code> value
97     * @see org.objectweb.cjdbc.controller.scheduler.schema.SchedulerDatabaseSchema
98     */
99    public synchronized void setDatabaseSchema(DatabaseSchema dbs)
100   {
101     if (schedulerDatabaseSchema == null)
102     {
103       logger.info("Setting new database schema");
104       schedulerDatabaseSchema = new SchedulerDatabaseSchema(dbs);
105     }
106     else
107     { // Schema is updated, compute the diff !
108       SchedulerDatabaseSchema newSchema = new SchedulerDatabaseSchema(dbs);
109       ArrayList   tables = schedulerDatabaseSchema.getTables();
110       ArrayList   newTables = newSchema.getTables();
111       if (newTables == null)
112       { // New schema is empty (no backend is active anymore)
113         logger.info("Removing all tables.");
114         schedulerDatabaseSchema = null;
115         return;
116       }
117 
118       // Remove extra-tables
119       for (int i = 0; i < tables.size(); i++)
120       {
121         SchedulerDatabaseTable t = (SchedulerDatabaseTable) tables.get(i);
122         if (!newSchema.hasTable(t.getName()))
123         {
124           schedulerDatabaseSchema.removeTable(t);
125           if (logger.isInfoEnabled())
126             logger.info("Removing table " + t.getName());
127         }
128       }
129 
130       // Add missing tables
131       int size = newTables.size();
132       for (int i = 0; i < size; i++)
133       {
134         SchedulerDatabaseTable t = (SchedulerDatabaseTable) newTables.get(i);
135         if (!schedulerDatabaseSchema.hasTable(t.getName()))
136         {
137           schedulerDatabaseSchema.addTable(t);
138           if (logger.isInfoEnabled())
139             logger.info("Adding table " + t.getName());
140         }
141       }
142     }
143   }
144 
145   /**
146    * Merge the given <code>DatabaseSchema</code> with the current one.
147    * 
148    * @param dbs a <code>DatabaseSchema</code> value
149    * @see org.objectweb.cjdbc.controller.scheduler.schema.SchedulerDatabaseSchema
150    */
151   public void mergeDatabaseSchema(DatabaseSchema dbs)
152   {
153     try
154     {
155       logger.info("Merging new database schema");
156       schedulerDatabaseSchema.mergeSchema(new SchedulerDatabaseSchema(dbs));
157     }
158     catch (Exception   e)
159     {
160       logger.error("Error while merging new database schema", e);
161     }
162   }
163 
164   /**
165    * Additionally to scheduling the request, this method replaces the SQL Date
166    * macros such as now() with the current date.
167    * 
168    * @see org.objectweb.cjdbc.controller.scheduler.AbstractScheduler#scheduleReadRequest(SelectRequest)
169    */
170   public final void scheduleReadRequest(SelectRequest request)
171       throws SQLException  
172   {
173     synchronized (this)
174     {
175       request.setId(requestId++);
176     }
177   }
178 
179   /**
180    * @see org.objectweb.cjdbc.controller.scheduler.AbstractScheduler#readCompletedNotify(SelectRequest)
181    */
182   public final void readCompletedNotify(SelectRequest request)
183   {
184   }
185 
186   /**
187    * Additionally to scheduling the request, this method replaces the SQL Date
188    * macros such as now() with the current date.
189    * 
190    * @see org.objectweb.cjdbc.controller.scheduler.AbstractScheduler#scheduleWriteRequest(AbstractWriteRequest)
191    */
192   public void scheduleNonSuspendedWriteRequest(AbstractWriteRequest request)
193       throws SQLException  , RollbackException
194   {
195     if (request.isCreate())
196     {
197       synchronized (this)
198       {
199         request.setId(requestId++);
200       }
201       return;
202     }
203 
204     SchedulerDatabaseTable t = schedulerDatabaseSchema.getTable(request
205         .getTableName());
206     if (t == null)
207     {
208       String   msg = "No table found for request " + request.getId();
209       logger.error(msg);
210       throw new SQLException  (msg);
211     }
212 
213     // Deadlock detection
214     TransactionExclusiveLock tableLock = t.getLock();
215     if (!request.isAutoCommit())
216     {
217       synchronized (this)
218       {
219         if (tableLock.isLocked())
220         { // Is the lock owner blocked by a lock we already own?
221           long owner = tableLock.getLocker();
222           long us = request.getTransactionId();
223           if (owner != us)
224           { // Parse all tables
225             ArrayList   tables = schedulerDatabaseSchema.getTables();
226             ArrayList   weAreblocking = new ArrayList  ();
227             int size = tables.size();
228             for (int i = 0; i < size; i++)
229             {
230               SchedulerDatabaseTable table = (SchedulerDatabaseTable) tables
231                   .get(i);
232               if (table == null)
233                 continue;
234               TransactionExclusiveLock lock = table.getLock();
235               // Are we the lock owner ?
236               if (lock.isLocked())
237               {
238                 if (lock.getLocker() == us)
239                 {
240                   // Is 'owner' in the list of the blocked transactions?
241                   if (lock.isWaiting(owner))
242                   { // Deadlock detected, we must rollback
243                     releaseLocks(us);
244                     throw new RollbackException(
245                         "Deadlock detected, rollbacking transaction " + us);
246                   }
247                   else
248                     weAreblocking.addAll(lock.getWaitingList());
249                 }
250               }
251             }
252           }
253           else
254           { // We are the lock owner and already synchronized on this
255             // Assign the request id and exit
256             request.setId(requestId++);
257             return;
258           }
259         }
260         else
261         { // Lock is free, take it in the synchronized block
262           acquireLockAndSetRequestId(request, tableLock);
263           return;
264         }
265       }
266     }
267 
268     acquireLockAndSetRequestId(request, tableLock);
269   }
270 
271   private void acquireLockAndSetRequestId(AbstractWriteRequest request,
272       TransactionExclusiveLock tableLock) throws SQLException  
273   {
274     // Acquire the lock
275     if (tableLock.acquire(request))
276     {
277       synchronized (this)
278       {
279         request.setId(requestId++);
280       }
281       if (logger.isDebugEnabled())
282         logger.debug("Request " + request.getId() + " scheduled for write ("
283             + getPendingWrites() + " pending writes)");
284     }
285     else
286     {
287       if (logger.isWarnEnabled())
288         logger.warn("Request " + request.getId() + " timed out ("
289             + request.getTimeout() + " ms)");
290       throw new SQLException  ("Timeout (" + request.getTimeout()
291           + ") for request: " + request.getId());
292     }
293   }
294 
295   /**
296    * @see org.objectweb.cjdbc.controller.scheduler.AbstractScheduler#notifyWriteCompleted(AbstractWriteRequest)
297    */
298   public final void notifyWriteCompleted(AbstractWriteRequest request)
299   {
300     if (request.isCreate())
301     { // Add table to schema
302       if (logger.isDebugEnabled())
303         logger.debug("Adding table '" + request.getTableName()
304             + "' to scheduler schema");
305       synchronized (this)
306       {
307         schedulerDatabaseSchema.addTable(new SchedulerDatabaseTable(
308             new DatabaseTable(request.getTableName())));
309       }
310     }
311     else if (request.isDrop())
312     { // Drop table from schema
313       if (logger.isDebugEnabled())
314         logger.debug("Removing table '" + request.getTableName()
315             + "' to scheduler schema");
316       synchronized (this)
317       {
318         schedulerDatabaseSchema.removeTable(schedulerDatabaseSchema
319             .getTable(request.getTableName()));
320       }
321       return;
322     }
323 
324     // Requests outside transaction delimiters must release the lock
325     // as soon as they have executed
326     if (request.isAutoCommit())
327     {
328       SchedulerDatabaseTable t = schedulerDatabaseSchema.getTable(request
329           .getTableName());
330       if (t == null)
331       {
332         String   msg = "No table found to release lock for request "
333             + request.getId();
334         logger.error(msg);
335       }
336       else
337         t.getLock().release();
338     }
339   }
340 
341   /**
342    * @see org.objectweb.cjdbc.controller.scheduler.AbstractScheduler#scheduleNonSuspendedStoredProcedure(org.objectweb.cjdbc.common.sql.StoredProcedure)
343    */
344   public void scheduleNonSuspendedStoredProcedure(StoredProcedure proc)
345       throws SQLException  , RollbackException
346   {
347     throw new SQLException  (
348         "Stored procedures are not supported by the RAIDb-1 optimistic transaction level scheduler.");
349   }
350 
351   /**
352    * @see org.objectweb.cjdbc.controller.scheduler.AbstractScheduler#notifyStoredProcedureCompleted(org.objectweb.cjdbc.common.sql.StoredProcedure)
353    */
354   public void notifyStoredProcedureCompleted(StoredProcedure proc)
355   {
356     // We should never execute here since scheduleNonSuspendedStoredProcedure
357     // should have failed prior calling us
358     throw new RuntimeException  (
359         "Stored procedures are not supported by the RAIDb-1 optimistic transaction level scheduler.");
360   }
361 
362   //
363   // Transaction Management
364   //
365 
366   /**
367    * @see org.objectweb.cjdbc.controller.scheduler.AbstractScheduler#commitTransaction(long)
368    */
369   protected final void commitTransaction(long transactionId)
370   {
371     releaseLocks(transactionId);
372   }
373 
374   /**
375    * @see org.objectweb.cjdbc.controller.scheduler.AbstractScheduler#rollbackTransaction(long)
376    */
377   protected final void rollbackTransaction(long transactionId)
378   {
379     releaseLocks(transactionId);
380   }
381 
382   /**
383    * @see org.objectweb.cjdbc.controller.scheduler.AbstractScheduler#rollbackTransaction(long,
384    *      String)
385    */
386   protected final void rollbackTransaction(long transactionId,
387       String   savepointName)
388   {
389   }
390 
391   /**
392    * @see org.objectweb.cjdbc.controller.scheduler.AbstractScheduler#setSavepointTransaction(long,
393    *      String)
394    */
395   protected final void setSavepointTransaction(long transactionId, String   name)
396   {
397   }
398 
399   /**
400    * @see org.objectweb.cjdbc.controller.scheduler.AbstractScheduler#releaseSavepointTransaction(long,
401    *      String)
402    */
403   protected final void releaseSavepointTransaction(long transactionId,
404       String   name)
405   {
406   }
407 
408   /**
409    * Release all locks we may own on tables.
410    * 
411    * @param transactionId id of the transaction that releases the locks
412    */
413   private synchronized void releaseLocks(long transactionId)
414   {
415     ArrayList   tables = schedulerDatabaseSchema.getTables();
416     int size = tables.size();
417     for (int i = 0; i < size; i++)
418     {
419       SchedulerDatabaseTable t = (SchedulerDatabaseTable) tables.get(i);
420       if (t == null)
421         continue;
422       TransactionExclusiveLock lock = t.getLock();
423       // Are we the lock owner ?
424       if (lock.isLocked())
425         if (lock.getLocker() == transactionId)
426           lock.release();
427     }
428   }
429 
430   //
431   // Debug/Monitoring
432   //
433 
434   /**
435    * @see org.objectweb.cjdbc.controller.scheduler.AbstractScheduler#getXmlImpl()
436    */
437   public String   getXmlImpl()
438   {
439     return "<" + DatabasesXmlTags.ELT_RAIDb1Scheduler + " "
440         + DatabasesXmlTags.ATT_level + "=\""
441         + DatabasesXmlTags.VAL_optimisticTransaction + "\"/>";
442   }
443 }
444
A to Z: JavaDoc & Examples Daily Java News & Articles Open Source Projects Open Source Codes Free Computer Books Remove Frame
Popular Tags