KickJava   Java API By Example, From Geeks To Geeks.

Java > Open Source Codes > org > objectweb > cjdbc > controller > scheduler > raidb2 > RAIDb2QueryLevelScheduler


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.raidb2;
26
27 import java.sql.SQLException JavaDoc;
28
29 import org.objectweb.cjdbc.common.exceptions.RollbackException;
30 import org.objectweb.cjdbc.common.sql.AbstractRequest;
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.xml.DatabasesXmlTags;
36 import org.objectweb.cjdbc.controller.requestmanager.RAIDbLevels;
37 import org.objectweb.cjdbc.controller.scheduler.AbstractScheduler;
38
39 /**
40  * This scheduler provides query level scheduling for RAIDb-2 controllers. Reads
41  * can execute in parallel until a write comes in. Then the write waits for the
42  * completion of the reads. Any new read is stacked after the write and they are
43  * released together when the write has completed its execution.
44  *
45  * @author <a HREF="mailto:Emmanuel.Cecchet@inria.fr">Emmanuel Cecchet </a>
46  * @author <a HREF="mailto:jbvanzuylen@transwide.com">Jean-Bernard van Zuylen
47  * </a>
48  * @version 1.0
49  */
50 public class RAIDb2QueryLevelScheduler extends AbstractScheduler
51 {
52
53   //
54 // How the code is organized ?
55 //
56 // 1. Member variables
57 // 2. Constructor
58 // 3. Request handling
59 // 4. Transaction management
60 // 5. Debug/Monitoring
61 //
62
63   private long requestId;
64   private int pendingReads;
65
66   // We have to distinguish read and write to wake up only
67 // waiting reads or writes according to the situation
68 private Object JavaDoc readSync; // to synchronize on reads completion
69 private Object JavaDoc writeSync; // to synchronize on writes completion
70
71   //
72 // Constructor
73 //
74
75   /**
76    * Creates a new Query Level Scheduler
77    */
78   public RAIDb2QueryLevelScheduler()
79   {
80     super(RAIDbLevels.RAIDb2, ParsingGranularities.NO_PARSING);
81     requestId = 0;
82     pendingReads = 0;
83     readSync = new Object JavaDoc();
84     writeSync = new Object JavaDoc();
85   }
86
87   //
88 // Request Handling
89 //
90
91   /**
92    * @see org.objectweb.cjdbc.controller.scheduler.AbstractScheduler#scheduleReadRequest(SelectRequest)
93    */
94   public void scheduleReadRequest(SelectRequest request) throws SQLException JavaDoc
95   {
96     // Now deal with synchronization
97 synchronized (this.writeSync)
98     {
99       if (getPendingWrites() == 0)
100       { // No writes pending, go ahead !
101 synchronized (this.readSync)
102         {
103           request.setId(requestId++);
104           pendingReads++;
105           if (logger.isDebugEnabled())
106             logger.debug("Request " + request.getId() + " scheduled for read ("
107                 + pendingReads + " pending reads)");
108           return;
109         }
110       }
111
112       // Wait for the writes completion
113 try
114       {
115         if (logger.isDebugEnabled())
116           logger.debug("Request " + request.getId() + " waiting for "
117               + getPendingWrites() + " pending writes)");
118
119         int timeout = request.getTimeout();
120         if (timeout > 0)
121         {
122           long start = System.currentTimeMillis();
123           // Convert seconds to milliseconds for wait call
124 long lTimeout = timeout * 1000;
125           this.writeSync.wait(lTimeout);
126           long end = System.currentTimeMillis();
127           int remaining = (int) (lTimeout - (end - start));
128           if (remaining > 0)
129             request.setTimeout(remaining);
130           else
131           {
132             String JavaDoc msg = "Timeout (" + request.getTimeout() + ") for request: "
133                 + request.getId();
134             logger.warn(msg);
135             throw new SQLException JavaDoc(msg);
136           }
137         }
138         else
139           this.writeSync.wait();
140
141         synchronized (this.readSync)
142         {
143           request.setId(requestId++);
144           pendingReads++;
145           if (logger.isDebugEnabled())
146             logger.debug("Request " + request.getId() + " scheduled for read ("
147                 + pendingReads + " pending reads)");
148           return; // Ok, write completed before timeout
149 }
150       }
151       catch (InterruptedException JavaDoc e)
152       {
153         // Timeout
154 if (logger.isWarnEnabled())
155           logger.warn("Request " + request.getId() + " timed out ("
156               + request.getTimeout() + " s)");
157         throw new SQLException JavaDoc("Timeout (" + request.getTimeout()
158             + ") for request: " + request.getId());
159       }
160     }
161   }
162
163   /**
164    * @see org.objectweb.cjdbc.controller.scheduler.AbstractScheduler#readCompletedNotify(SelectRequest)
165    */
166   public final void readCompletedNotify(SelectRequest request)
167   {
168     synchronized (this.readSync)
169     {
170       pendingReads--;
171       if (logger.isDebugEnabled())
172         logger.debug("Request " + request.getId() + " completed");
173       if (pendingReads == 0)
174       {
175         if (logger.isDebugEnabled())
176           logger.debug("Last read completed, notifying writes");
177         readSync.notifyAll(); // Wakes up any waiting write query
178 }
179     }
180   }
181
182   /**
183    * @see org.objectweb.cjdbc.controller.scheduler.AbstractScheduler#scheduleWriteRequest(AbstractWriteRequest)
184    */
185   public void scheduleNonSuspendedWriteRequest(AbstractWriteRequest request)
186       throws SQLException JavaDoc
187   {
188     // We have to take the locks in the same order as reads else
189 // we could have a deadlock
190 synchronized (this.writeSync)
191     {
192       synchronized (this.readSync)
193       {
194         if (pendingReads == 0)
195         { // No read pending, go ahead
196 request.setId(requestId++);
197           if (logger.isDebugEnabled())
198             logger.debug("Request " + request.getId()
199                 + " scheduled for write (" + getPendingWrites()
200                 + " pending writes)");
201           return;
202         }
203       }
204     }
205
206     waitForReadCompletion(request);
207     scheduleNonSuspendedWriteRequest(request);
208   }
209
210   /**
211    * @see org.objectweb.cjdbc.controller.scheduler.AbstractScheduler#notifyWriteCompleted(AbstractWriteRequest)
212    */
213   public final synchronized void notifyWriteCompleted(
214       AbstractWriteRequest request)
215   {
216     synchronized (this.writeSync)
217     {
218       if (logger.isDebugEnabled())
219         logger.debug("Request " + request.getId() + " completed");
220       if (getPendingWrites() == 0)
221       {
222         if (logger.isDebugEnabled())
223           logger.debug("Last write completed, notifying reads");
224         writeSync.notifyAll(); // Wakes up all waiting read queries
225 }
226     }
227   }
228
229   /**
230    * @see org.objectweb.cjdbc.controller.scheduler.AbstractScheduler#scheduleNonSuspendedStoredProcedure(org.objectweb.cjdbc.common.sql.StoredProcedure)
231    */
232   public final synchronized void scheduleNonSuspendedStoredProcedure(
233       StoredProcedure proc) throws SQLException JavaDoc, RollbackException
234   {
235     // We have to take the locks in the same order as reads else
236 // we could have a deadlock
237 synchronized (this.writeSync)
238     {
239       synchronized (this.readSync)
240       {
241         if (pendingReads == 0)
242         { // No read pending, go ahead
243 proc.setId(requestId++);
244           if (logger.isDebugEnabled())
245             logger.debug("Stored procedure "
246                 + proc.getId()
247                 + (proc.isAutoCommit() ? "" : " transaction "
248                     + proc.getTransactionId()) + " scheduled for write ("
249                 + getPendingWrites() + " pending writes)");
250           return;
251         }
252       }
253     }
254
255     waitForReadCompletion(proc);
256     scheduleNonSuspendedStoredProcedure(proc);
257   }
258
259   /**
260    * @see org.objectweb.cjdbc.controller.scheduler.AbstractScheduler#notifyStoredProcedureCompleted(org.objectweb.cjdbc.common.sql.StoredProcedure)
261    */
262   public final void notifyStoredProcedureCompleted(StoredProcedure proc)
263   {
264     synchronized (this.writeSync)
265     {
266       if (logger.isDebugEnabled())
267         logger.debug("Stored procedure " + proc.getId() + " completed - "
268             + getPendingWrites() + " pending writes");
269       if (getPendingWrites() == 0)
270       {
271         if (logger.isDebugEnabled())
272           logger.debug("Last write completed, notifying reads");
273         writeSync.notifyAll(); // Wakes up all waiting read queries
274 }
275     }
276   }
277
278   /**
279    * Wait for the reads completion. Synchronizes on this.readSync.
280    *
281    * @param request the request that is being scheduled
282    * @throws SQLException if an error occurs
283    */
284   private void waitForReadCompletion(AbstractRequest request)
285       throws SQLException JavaDoc
286   {
287     synchronized (this.readSync)
288     {
289       // Wait for the reads completion
290 try
291       {
292         if (logger.isDebugEnabled())
293           logger.debug("Request " + request.getId() + " waiting for "
294               + pendingReads + " pending reads)");
295
296         int timeout = request.getTimeout();
297         if (timeout > 0)
298         {
299           long start = System.currentTimeMillis();
300           // Convert seconds to milliseconds for wait call
301 long lTimeout = timeout * 1000;
302           this.readSync.wait(lTimeout);
303           long end = System.currentTimeMillis();
304           int remaining = (int) (lTimeout - (end - start));
305           if (remaining > 0)
306             request.setTimeout(remaining);
307           else
308           {
309             String JavaDoc msg = "Timeout (" + request.getTimeout() + ") for request: "
310                 + request.getId();
311             logger.warn(msg);
312             throw new SQLException JavaDoc(msg);
313           }
314         }
315         else
316           this.readSync.wait();
317       }
318       catch (InterruptedException JavaDoc e)
319       {
320         // Timeout
321 if (logger.isWarnEnabled())
322           logger.warn("Request " + request.getId() + " timed out ("
323               + request.getTimeout() + " ms)");
324         throw new SQLException JavaDoc("Timeout (" + request.getTimeout()
325             + ") for request: " + request.getId());
326       }
327     }
328   }
329
330   //
331 // Transaction Management
332 //
333
334   /**
335    * @see org.objectweb.cjdbc.controller.scheduler.AbstractScheduler#commitTransaction(long)
336    */
337   protected final void commitTransaction(long transactionId)
338   {
339   }
340
341   /**
342    * @see org.objectweb.cjdbc.controller.scheduler.AbstractScheduler#rollbackTransaction(long)
343    */
344   protected final void rollbackTransaction(long transactionId)
345   {
346   }
347
348   /**
349    * @see org.objectweb.cjdbc.controller.scheduler.AbstractScheduler#rollbackTransaction(long,
350    * String)
351    */
352   protected final void rollbackTransaction(long transactionId,
353       String JavaDoc savepointName)
354   {
355   }
356
357   /**
358    * @see org.objectweb.cjdbc.controller.scheduler.AbstractScheduler#setSavepointTransaction(long,
359    * String)
360    */
361   protected final void setSavepointTransaction(long transactionId, String JavaDoc name)
362   {
363   }
364
365   /**
366    * @see org.objectweb.cjdbc.controller.scheduler.AbstractScheduler#releaseSavepointTransaction(long,
367    * String)
368    */
369   protected final void releaseSavepointTransaction(long transactionId,
370       String JavaDoc name)
371   {
372   }
373
374   //
375 // Debug/Monitoring
376 //
377 /**
378    * @see org.objectweb.cjdbc.controller.scheduler.AbstractScheduler#getXmlImpl()
379    */
380   public String JavaDoc getXmlImpl()
381   {
382     StringBuffer JavaDoc info = new StringBuffer JavaDoc();
383     info.append("<" + DatabasesXmlTags.ELT_RAIDb2Scheduler + " "
384         + DatabasesXmlTags.ATT_level + "=\"" + DatabasesXmlTags.VAL_query
385         + "\"/>");
386     info.append(System.getProperty("line.separator"));
387     return info.toString();
388   }
389 }
390
Popular Tags