WriteRequestTask


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): Julie Marguerite, Jaco Swart.
23   */
24  
25  package org.objectweb.cjdbc.controller.loadbalancer.tasks;
26  
27  import java.sql.Connection  ;
28  import java.sql.SQLException  ;
29  
30  import org.objectweb.cjdbc.common.exceptions.NoTransactionStartWhenDisablingException;
31  import org.objectweb.cjdbc.common.exceptions.UnreachableBackendException;
32  import org.objectweb.cjdbc.common.log.Trace;
33  import org.objectweb.cjdbc.common.sql.AbstractWriteRequest;
34  import org.objectweb.cjdbc.controller.backend.DatabaseBackend;
35  import org.objectweb.cjdbc.controller.connection.AbstractConnectionManager;
36  import org.objectweb.cjdbc.controller.loadbalancer.AbstractLoadBalancer;
37  import org.objectweb.cjdbc.controller.loadbalancer.BackendWorkerThread;
38  
39  /**
40   * Executes an <code>AbstractWriteRequest</code> statement.
41   * 
42   * @author <a HREF="mailto:Emmanuel.Cecchet@inria.fr">Emmanuel Cecchet </a>
43   * @author <a HREF="mailto:Julie.Marguerite@inria.fr">Julie Marguerite </a>
44   * @author <a HREF="mailto:jaco.swart@iblocks.co.uk">Jaco Swart </a>
45   * @version 1.0
46   */
47  public class WriteRequestTask extends AbstractTask
48  {
49    private AbstractWriteRequest request;
50    private int                  result;
51  
52    /**
53     * Creates a new <code>WriteRequestTask</code>.
54     * 
55     * @param nbToComplete number of threads that must succeed before returning
56     * @param totalNb total number of threads
57     * @param request an <code>AbstractWriteRequest</code>
58     */
59    public WriteRequestTask(int nbToComplete, int totalNb,
60        AbstractWriteRequest request)
61    {
62      super(nbToComplete, totalNb);
63      this.request = request;
64    }
65  
66    /**
67     * Executes a write request with the given backend thread.
68     * 
69     * @param backendThread the backend thread that will execute the task
70     * @throws SQLException if an error occurs
71     */
72    public void executeTask(BackendWorkerThread backendThread)
73        throws SQLException  
74    {
75      DatabaseBackend backend = backendThread.getBackend();
76  
77      AbstractConnectionManager cm = backend.getConnectionManager(request
78          .getLogin());
79      if (cm == null)
80      {
81        SQLException   se = new SQLException  (
82            "No Connection Manager for Virtual Login:" + request.getLogin());
83        try
84        {
85          notifyFailure(backendThread, 1, se);
86        }
87        catch (SQLException   ignore)
88        {
89        }
90        throw se;
91      }
92  
93      Trace logger = backendThread.getLogger();
94      if (request.isAutoCommit())
95      {
96        if (backend.isDisabling())
97        {
98          // Backend is disabling, we do not execute queries except the one in the
99          // transaction we already started. Just notify the completion for the
100         // others.
101         notifyCompletion();
102         return;
103       }
104 
105       // Use a connection just for this request
106       Connection   c = null;
107       try
108       {
109         c = cm.getConnection();
110       }
111       catch (UnreachableBackendException e1)
112       {
113         SQLException   se = new SQLException  ("Backend " + backend.getName()
114             + " is no more reachable.");
115         try
116         {
117           notifyFailure(backendThread, 1, se);
118         }
119         catch (SQLException   ignore)
120         {
121         }
122         // Disable this backend (it is no more in sync) by killing the backend
123         // thread
124         backendThread.kill();
125         logger.error("Disabling backend " + backend.getName()
126             + " because it is no more reachable.");
127         throw se;
128       }
129 
130       // Sanity check
131       if (c == null)
132       {
133         SQLException   se = new SQLException  ("No more connections");
134         try
135         { // All backends failed, just ignore
136           if (!notifyFailure(backendThread, (long) request.getTimeout() * 1000,
137               se))
138           {
139             return;
140           }
141         }
142         catch (SQLException   ignore)
143         {
144         }
145         // Disable this backend (it is no more in sync) by killing the backend
146         // thread
147         backendThread.kill();
148         String   msg = "Request '"
149             + request.getSQLShortForm(backend.getSQLShortFormLength())
150             + "' failed on backend " + backend.getName() + " but "
151             + getSuccess() + " succeeded (" + se + ")";
152         logger.error(msg);
153         throw new SQLException  (msg);
154       }
155 
156       // Execute Query
157       try
158       {
159         result = AbstractLoadBalancer.executeUpdateRequestOnBackend(request,
160             backend, c);
161 
162         backend.updateDatabaseBackendSchema(request);
163       }
164       catch (Exception   e)
165       {
166         try
167         { // All backends failed, just ignore
168           if (!notifyFailure(backendThread, (long) request.getTimeout() * 1000,
169               e))
170             return;
171         }
172         catch (SQLException   ignore)
173         {
174         }
175         // Disable this backend (it is no more in sync) by killing the backend
176         // thread
177         backendThread.kill();
178         String   msg = "Request '"
179             + request.getSQLShortForm(backend.getSQLShortFormLength())
180             + "' failed on backend " + backend.getName() + " but "
181             + getSuccess() + " succeeded (" + e + ")";
182 
183         if (logger.isDebugEnabled())
184           logger.debug(msg, e);
185         else
186           logger.error(msg);
187         throw new SQLException  (msg);
188       }
189       finally
190       {
191         cm.releaseConnection(c);
192       }
193     }
194     else
195     { // Re-use the connection used by this transaction
196       Connection   c;
197       long tid = request.getTransactionId();
198       Long   lTid = new Long  (tid);
199 
200       try
201       {
202         c = backend.getConnectionForTransactionAndLazyBeginIfNeeded(lTid, cm,
203             request.getTransactionIsolation());
204       }
205       catch (UnreachableBackendException ube)
206       {
207         SQLException   se = new SQLException  ("Backend " + backend.getName()
208             + " is no more reachable.");
209         try
210         {
211           notifyFailure(backendThread, 1, se);
212         }
213         catch (SQLException   ignore)
214         {
215         }
216         // Disable this backend (it is no more in sync) by killing the backend
217         // thread
218         backendThread.kill();
219         logger.error("Disabling backend " + backend.getName()
220             + " because it is no more reachable.");
221         throw se;
222       }
223       catch (NoTransactionStartWhenDisablingException e)
224       {
225         // Backend is disabling, we do not execute queries except the one in the
226         // transaction we already started. Just notify the completion for the
227         // others.
228         notifyCompletion();
229         return;
230       }
231       catch (SQLException   e1)
232       {
233         SQLException   se = new SQLException  (
234             "Unable to get connection for transaction " + tid);
235         try
236         { // All backends failed, just ignore
237           if (!notifyFailure(backendThread, (long) request.getTimeout() * 1000,
238               se))
239             return;
240         }
241         catch (SQLException   ignore)
242         {
243         }
244         // Disable this backend (it is no more in sync) by killing the
245         // backend thread
246         backendThread.kill();
247         String   msg = "Request '"
248             + request.getSQLShortForm(backend.getSQLShortFormLength())
249             + "' failed on backend " + backend.getName() + " but "
250             + getSuccess() + " succeeded (" + se + ")";
251         logger.error(msg);
252         throw new SQLException  (msg);
253       }
254 
255       // Sanity check
256       if (c == null)
257       { // Bad connection
258         SQLException   se = new SQLException  (
259             "Unable to retrieve connection for transaction " + tid);
260         try
261         { // All backends failed, just ignore
262           if (!notifyFailure(backendThread, (long) request.getTimeout() * 1000,
263               se))
264             return;
265         }
266         catch (SQLException   ignore)
267         {
268         }
269         // Disable this backend (it is no more in sync) by killing the
270         // backend thread
271         backendThread.kill();
272         String   msg = "Request '"
273             + request.getSQLShortForm(backend.getSQLShortFormLength())
274             + "' failed on backend " + backend.getName() + " but "
275             + getSuccess() + " succeeded (" + se + ")";
276         logger.error(msg);
277         throw new SQLException  (msg);
278       }
279 
280       // Execute Query
281       try
282       {
283         result = AbstractLoadBalancer.executeUpdateRequestOnBackend(request,
284             backend, c);
285 
286         backend.updateDatabaseBackendSchema(request);
287       }
288       catch (Exception   e)
289       {
290         try
291         { // All backends failed, just ignore
292           if (!notifyFailure(backendThread, (long) request.getTimeout() * 1000,
293               e))
294             return;
295         }
296         catch (SQLException   ignore)
297         {
298         }
299         // Disable this backend (it is no more in sync) by killing the backend
300         // thread
301         backendThread.kill();
302         String   msg = "Request '"
303             + request.getSQLShortForm(backend.getSQLShortFormLength())
304             + "' failed on backend " + backend.getName() + " but "
305             + getSuccess() + " succeeded (" + e + ")";
306         if (logger.isDebugEnabled())
307           logger.debug(msg, e);
308         else
309           logger.error(msg);
310         throw new SQLException  (msg);
311       }
312     }
313     notifySuccess();
314   }
315 
316   /**
317    * Returns the result.
318    * 
319    * @return int
320    */
321   public int getResult()
322   {
323     return result;
324   }
325 
326   /**
327    * @see java.lang.Object#toString()
328    */
329   public String   toString()
330   {
331     if (request.isAutoCommit())
332       return "WriteAutocommit Task (" + request.getSQL() + ")";
333     else
334       return "Write Task from transaction:" + request.getTransactionId() + "("
335           + request.getSQL() + ")";
336   }
337 
338 }
A to Z: JavaDoc & Examples Daily Java News & Articles Open Source Projects Open Source Codes Free Computer Books Remove Frame
Popular Tags