KickJava   Java API By Example, From Geeks To Geeks.

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


1 /**
2  * Sequoia: Database clustering technology.
3  * Copyright (C) 2002-2004 French National Institute For Research In Computer
4  * Science And Control (INRIA).
5  * Contact: sequoia@continuent.org
6  *
7  * Licensed under the Apache License, Version 2.0 (the "License");
8  * you may not use this file except in compliance with the License.
9  * You may obtain a copy of the License at
10  *
11  * http://www.apache.org/licenses/LICENSE-2.0
12  *
13  * Unless required by applicable law or agreed to in writing, software
14  * distributed under the License is distributed on an "AS IS" BASIS,
15  * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
16  * See the License for the specific language governing permissions and
17  * limitations under the License.
18  *
19  * Initial developer(s): Emmanuel Cecchet.
20  * Contributor(s):
21  */

22
23 package org.continuent.sequoia.controller.loadbalancer;
24
25 import java.sql.SQLException JavaDoc;
26 import java.sql.Statement JavaDoc;
27
28 import org.continuent.sequoia.common.i18n.Translate;
29 import org.continuent.sequoia.common.log.Trace;
30 import org.continuent.sequoia.controller.backend.DatabaseBackend;
31
32 /**
33  * Process sequentially a set of tasks and send them to a backend.
34  *
35  * @author <a HREF="mailto:Emmanuel.Cecchet@inria.fr">Emmanuel Cecchet </a>
36  * @version 1.0
37  */

38 public class BackendWorkerThread extends Thread JavaDoc
39 {
40   //
41
// How the code is organized ?
42
// 1. Member variables
43
// 2. Constructor(s)
44
// 3. Task management
45
// 4. Getter/Setters
46
//
47

48   private AbstractLoadBalancer loadBalancer;
49   private DatabaseBackend backend;
50   private boolean isKilled = false;
51
52   private Trace logger = null;
53   private BackendTaskQueues queues;
54   private Statement JavaDoc currentStatement;
55
56   /** True if this thread only plays commit/rollback tasks */
57   private boolean playingCommitRollbackOnly = false;
58
59   /*
60    * Constructor
61    */

62
63   /**
64    * Creates a new <code>BackendWorkerThread</code>.
65    *
66    * @param backend the backend this thread is associated to.
67    * @param loadBalancer the load balancer instanciating this thread
68    */

69   public BackendWorkerThread(DatabaseBackend backend,
70       AbstractLoadBalancer loadBalancer)
71   {
72     this(loadBalancer.vdb.getVirtualDatabaseName()
73         + " - BackendWorkerThread for backend '" + backend.getName()
74         + "' with RAIDb level:" + loadBalancer.getRAIDbLevel(), backend,
75         loadBalancer);
76   }
77
78   /**
79    * Creates a new <code>BackendWorkerThread</code>.
80    *
81    * @param name the name to give to the thread
82    * @param backend the backend this thread is associated to.
83    * @param loadBalancer the load balancer instanciating this thread
84    */

85   public BackendWorkerThread(String JavaDoc name, DatabaseBackend backend,
86       AbstractLoadBalancer loadBalancer)
87   {
88     super(name);
89     // Sanity checks
90
if (backend == null)
91     {
92       String JavaDoc msg = Translate.get("backendworkerthread.null.backend");
93       logger = Trace
94           .getLogger("org.continuent.sequoia.controller.backend.DatabaseBackend");
95       logger.error(msg);
96       throw new NullPointerException JavaDoc(msg);
97     }
98
99     logger = Trace
100         .getLogger("org.continuent.sequoia.controller.backend.DatabaseBackend."
101             + backend.getName());
102
103     if (loadBalancer == null)
104     {
105       String JavaDoc msg = Translate.get("backendworkerthread.null.loadbalancer");
106       logger.error(msg);
107       throw new NullPointerException JavaDoc(msg);
108     }
109
110     this.backend = backend;
111     this.loadBalancer = loadBalancer;
112     this.queues = backend.getTaskQueues();
113   }
114
115   /**
116    * Kills this thread after the next task processing and forces the load
117    * balancer to disable the backend. It also marks all remaining tasks in the
118    * task list as failed.
119    */

120   public void killAndForceDisable()
121   {
122     kill(true);
123   }
124
125   /**
126    * Kills this thread after the next task processing. It also disables the
127    * backend.
128    */

129   public void killWithoutDisablingBackend()
130   {
131     kill(false);
132   }
133
134   /**
135    * Kills this thread after the next task processing. It also marks all
136    * remaining tasks in the task list as failed.
137    *
138    * @param forceDisable true if the task must call the load balancer to disable
139    * the backend
140    */

141   private void kill(boolean forceDisable)
142   {
143     synchronized (this)
144     {
145       if (logger.isDebugEnabled())
146         logger.debug(this.getName() + " is shutting down");
147
148       isKilled = true;
149       notify(); // Wake up thread
150
}
151     if (forceDisable)
152     {
153       try
154       {
155         // This ensure that all worker threads get removed from the load
156
// balancer list and that the backend state is set to disable.
157
loadBalancer.disableBackend(backend, true);
158       }
159       catch (SQLException JavaDoc ignore)
160       {
161       }
162     }
163   }
164
165   /**
166    * Process the tasklist and call <code>wait()</code> (on itself) when the
167    * tasklist becomes empty.
168    */

169   public void run()
170   {
171     BackendTaskQueueEntry currentlyProcessingEntry = null;
172
173     while (!isKilled)
174     {
175       try
176       {
177         // Take the next available task
178
if (isPlayingCommitRollbackOnly())
179           currentlyProcessingEntry = queues
180               .getNextCommitRollbackToExecute(this);
181         else
182           currentlyProcessingEntry = queues.getNextEntryToExecute(this);
183
184         if (currentlyProcessingEntry == null)
185         {
186           logger.warn("Null task in BackendWorkerThread");
187           continue;
188         }
189
190         // Execute the tasks
191
if (logger.isDebugEnabled())
192           logger.debug(Translate.get("backendworkerthread.execute.task",
193               currentlyProcessingEntry.getTask().toString()));
194         currentlyProcessingEntry.getTask().execute(this);
195       }
196       catch (SQLException JavaDoc e)
197       {
198         // Task should have notified of failure
199
logger.warn(Translate.get("backendworkerthread.task.failed", e));
200       }
201       catch (RuntimeException JavaDoc re)
202       {
203         logger.fatal(Translate.get(
204             "backendworkerthread.task.runtime.exception",
205             currentlyProcessingEntry == null
206                 ? "null task"
207                 : currentlyProcessingEntry.toString()), re);
208
209         // We can't know for sure if the task has notified the failure or not.
210
// To prevent a deadlock, we force the failure notification here.
211
try
212         {
213           currentlyProcessingEntry.getTask().notifyFailure(this, 1,
214               new SQLException JavaDoc(re.getMessage()));
215         }
216         catch (SQLException JavaDoc e1)
217         {
218           // just notify
219
}
220       }
221       finally
222       {
223         setCurrentStatement(null);
224         try
225         {
226           if (currentlyProcessingEntry != null)
227           {
228             queues.completedEntryExecution(currentlyProcessingEntry);
229             if (logger.isDebugEnabled())
230               logger.debug(Translate.get(
231                   "backendworkerthread.execute.task.completed",
232                   currentlyProcessingEntry.getTask().toString()));
233           }
234         }
235         catch (RuntimeException JavaDoc e)
236         {
237           logger.warn(
238               Translate.get("backendworkerthread.remove.task.error", e), e);
239         }
240       }
241     } // end while (!isKilled)
242
}
243
244   /*
245    * Getter/Setter
246    */

247
248   /**
249    * Returns the backend.
250    *
251    * @return a <code>DatabaseBackend</code> instance
252    */

253   public DatabaseBackend getBackend()
254   {
255     return backend;
256   }
257
258   /**
259    * Returns the currentStatement value.
260    *
261    * @return Returns the currentStatement.
262    */

263   public final Statement JavaDoc getCurrentStatement()
264   {
265     return currentStatement;
266   }
267
268   /**
269    * Sets the currentStatement value.
270    *
271    * @param currentStatement The currentStatement to set.
272    */

273   public final void setCurrentStatement(Statement JavaDoc currentStatement)
274   {
275     this.currentStatement = currentStatement;
276   }
277
278   /**
279    * Returns the loadBalancer value.
280    *
281    * @return Returns the loadBalancer.
282    */

283   public final AbstractLoadBalancer getLoadBalancer()
284   {
285     return loadBalancer;
286   }
287
288   /**
289    * Returns the logger for tracing.
290    *
291    * @return a <code>Trace</code> instance
292    */

293   public Trace getLogger()
294   {
295     return logger;
296   }
297
298   /**
299    * Returns the playCommitRollbackOnly value.
300    *
301    * @return Returns the playCommitRollbackOnly.
302    */

303   public boolean isPlayingCommitRollbackOnly()
304   {
305     return playingCommitRollbackOnly;
306   }
307
308   /**
309    * Sets the playCommitRollbackOnly value.
310    *
311    * @param playCommitRollbackOnly The playCommitRollbackOnly to set.
312    */

313   public void setPlayCommitRollbackOnly(boolean playCommitRollbackOnly)
314   {
315     this.playingCommitRollbackOnly = playCommitRollbackOnly;
316   }
317
318   /**
319    * @see java.lang.Object#equals(java.lang.Object)
320    */

321   public boolean equals(Object JavaDoc obj)
322   {
323     if (obj instanceof BackendWorkerThread)
324     {
325       return backend.equals(((BackendWorkerThread) obj).getBackend());
326     }
327     return false;
328   }
329
330   /**
331    * @see java.lang.Object#hashCode()
332    */

333   public int hashCode()
334   {
335     return backend.hashCode();
336   }
337
338 }
Popular Tags