MultiPool


1    /*
2   
3   This software is OSI Certified Open Source Software.
4   OSI Certified is a certification mark of the Open Source Initiative.
5   
6   The license (Mozilla version 1.0) can be read at the MMBase site.
7   See http://www.MMBase.org/license
8   
9    */
10  package org.mmbase.module.database;
11  
12  import java.sql.*;
13  import java.util.*;
14  import org.mmbase.util.DijkstraSemaphore;
15  import org.mmbase.module.core.MMBase;
16  import org.mmbase.util.logging.Logger;
17  import org.mmbase.util.logging.Logging;
18  
19  /**
20   * JDBC Pool, a dummy interface to multiple real connection
21   * @javadoc
22   * @author vpro
23   * @version $Id: MultiPool.java,v 1.57 2005/12/17 16:18:58 michiel Exp $
24   */
25  public class MultiPool {
26  
27      private static final Logger log    = Logging.getLoggerInstance(MultiPool.class);
28  
29      private List              pool     = null;
30      private List              busyPool = new ArrayList();
31      private int               conMax   = 4;
32      private DijkstraSemaphore semaphore = null;
33      private int      totalConnections = 0;
34      private int      maxQueries = 500;
35      private String     name;
36      private String     password;
37      private String     url;
38      private DatabaseSupport databaseSupport;
39  
40      private boolean doReconnect  = true;
41  
42      private long    maxLifeTime  = 120000;
43      private long    maxZeroTime  = maxLifeTime / 4;
44  
45      /**
46       * @javadoc
47       */
48      MultiPool(DatabaseSupport databaseSupport, String   url, String   name, String   password, int conMax) throws SQLException {
49          this(databaseSupport, url, name, password, conMax, 500);
50  
51      }
52      /**
53       * Establish connection to the JDBC Pool(s)
54       */
55      MultiPool(DatabaseSupport databaseSupport, String   url, String   name, String   password, int conMax,int maxQueries) throws SQLException {
56  
57          this.conMax          = conMax;
58          this.name            = name;
59          this.url             = url;
60          this.password        = password;
61          this.maxQueries      = maxQueries;
62          if (this.maxQueries <= 0) doReconnect = false;
63          this.databaseSupport = databaseSupport;
64          log.service("Creating a multipool for database " + name + "(" + url + ") containing : " + conMax + " connections, " + (doReconnect ? "which will be refreshed after " + this.maxQueries + " queries"  : "which will not be refreshed"));
65          createPool();
66      }
67  
68      /**
69       * Set the time in ms how long a query may live before it is killed.
70       * @since MMBase-1.8
71       */
72      void setMaxLifeTime(long maxLifeTime) {
73          this.maxLifeTime = maxLifeTime;
74          maxZeroTime = maxLifeTime / 4;
75      }
76  
77      /**
78       * Gets the time in ms how long a query may live before it is killed.
79       * @since MMBase-1.8
80       */
81      long getMaxLifeTime() {
82          return maxLifeTime;
83      }
84  
85      /**
86       * Creates and fills the connection pool
87       * @since MMBase-1.7
88      */
89      protected void createPool() {
90          pool = new ArrayList();
91          MMBase mmb = MMBase.getMMBase();
92          boolean logStack = true;
93          try {
94              while (!fillPool(logStack)) {
95                  log.error("Cannot run with no connections, retrying after 10 seconds for " + mmb + " " + (mmb.isShutdown() ? "(shutdown)" : ""));
96                  Thread.sleep(10000);
97                  logStack = false; // don't log that mess a second time
98                  if (mmb.isShutdown()) {
99                      log.info("MMBase has been shutted down.");
100                     return;
101                 }
102             }
103             semaphore = new DijkstraSemaphore(pool.size());
104             log.service("Connection pool has " + conMax + " connections");
105         } catch (InterruptedException   ie) {
106             log.info("Interrupted: " + ie.getMessage());
107         }
108 
109 
110     }
111 
112 
113     /**
114      * Fills the connection pool.
115      * @return true if the pool contains connections and mmbase can be started.
116      * @since MMBase-1.7
117      */
118     protected boolean fillPool(boolean logStack) {
119         int errors = 0;
120         SQLException firstError = null;
121         // put connections on the pool
122         for (int i = 0; i < conMax ; i++) {
123             try {
124                 pool.add(getMultiConnection());
125             } catch (SQLException se) {
126                 errors++;
127                 if (log.isDebugEnabled()) {
128                     log.debug("i: " + "error " + errors + ": " + se.getMessage());
129                 }
130                 if (firstError == null) {
131                     firstError = se;
132                 }
133             }
134         }
135         if (errors > 0) {
136             String   message = firstError.getMessage();
137             if (logStack) {
138                 message += " " + Logging.stackTrace(firstError);
139             }  else {
140                 int nl = message.indexOf('\n'); // some stupid drivers (postgresql) add stacktrace to message
141                 if (nl > 0) {
142                     message = message.substring(0, nl) + "..."; // take most of it away again.
143                 }
144             }
145 
146             log.error("Could not get all connections (" + errors + " failures, multipool size now " + pool.size() + " rather then " + conMax +"). First error: " + message);
147             if (pool.size() < 1) { // that is fatal.
148                 return false;
149             }
150             this.conMax = pool.size();
151         }
152 
153         return true;
154 
155 
156     }
157 
158     /**
159      * Request a new 'real' Connection and wraps it in a new 'MultiConnection' object.
160      *
161      * @since MMBase-1.7
162      */
163     protected MultiConnection getMultiConnection() throws SQLException {
164         log.debug("Getting a new connection for url '" + url + "'");
165         Connection con;
166         if (name.equals("url") && password.equals("url")) {
167             con = DriverManager.getConnection(url);
168         } else {
169             con = DriverManager.getConnection(url, name, password);
170         }
171         databaseSupport.initConnection(con);
172         return new MultiConnection(this, con);
173     }
174 
175     /**
176      * Tries to fix this multi-connection if it is broken (e.g. if database restarted).
177      * @since MMBase-1.7.1
178      */
179     protected void replaceConnection(MultiConnection multiCon) throws SQLException {
180         if (name.equals("url") && password.equals("url")) {
181             multiCon.con = DriverManager.getConnection(url);
182         } else {
183             multiCon.con = DriverManager.getConnection(url, name, password);
184         }
185         databaseSupport.initConnection(multiCon.con);
186 
187     }
188 
189     protected void finalize() {
190         shutdown();
191     }
192 
193     /**
194      * 'realcloses' all connections.
195      * @since MMBase-1.6.2
196      */
197     public void shutdown() {
198         log.info("Shutting down multipool " + this);
199         if (semaphore == null) return; // nothing to shut down
200         synchronized (semaphore) {
201             try {
202                 for (Iterator i = busyPool.iterator(); i.hasNext();) {
203                     MultiConnection con = (MultiConnection) i.next();
204                     con.realclose();
205                 }
206                 busyPool.clear();
207                 for (Iterator i = pool.iterator(); i.hasNext();) {
208                     MultiConnection con = (MultiConnection) i.next();
209                     con.realclose();
210                 }
211                 pool.clear();
212             } catch (Throwable   e) {
213                 log.error(e);
214             } finally {
215                 conMax = busyPool.size() + pool.size(); // should be 0 now
216                 log.info("Having " + conMax + " connections now.");
217                 if (conMax != 0) log.error("Still having connections!");
218             }
219         }
220     }
221 
222     /**
223      * Check the connections
224      * @bad-constant  Max life-time of a query must be configurable
225      */
226     void checkTime() {
227 
228         if (log.isDebugEnabled()) {
229             log.debug("JDBC -> Starting the pool check (" + this + ") : busy=" + busyPool.size() + " free=" + pool.size());
230         }
231         if (semaphore == null ||  // during start-up this could happen.
232             conMax == 0           // during shut-down this could happen
233             ) return;
234         synchronized (semaphore) {
235 
236             int releaseCount = 0; // number of connection that are put back to pool
237 
238             //lock semaphore, so during the checks, no connections can be acquired or put back
239 
240             //// (because the methods of semaphore are synchronized)
241             //// commented out above commentline, because this is not true.
242             //// The synchronized in java works on instnaces of a class.
243             //// The code
244             ////
245             ////   void synchronized myMethod() ( statement; )
246             ////
247             //// can be rewritten as
248             ////
249             ////   void myMethod(){
250             ////       synchronized(this) { statement; }
251             ////   }
252             ////
253             //// This way the instance is only locked when the method is executing.
254             //// The statements before and after this are not synchronized.
255             //// When an instnace is not in a synchronized block the instance is not locked
256             //// and every thread can modify the instance.
257             //// Busypool is in this method not locked and can be modified in getFree and putBack.
258             //// if the busypool is iterated and modified at the same time a
259             //// ConcurrentModificationException is thrown.
260 
261             // Michiel: but the getFree and putBack actions on the two pools are also synchronized on semaphore.
262             //          so nothing can edit them without having acquired the lock on semaphore.
263 
264 
265             long nowTime = System.currentTimeMillis();
266 
267             for (Iterator i = busyPool.iterator(); i.hasNext();) {
268                 MultiConnection con = (MultiConnection) i.next();
269 
270                 boolean isClosed = true;
271 
272                 try {
273                     isClosed = con.isClosed();
274                 } catch (SQLException e) {
275                     log.warn("Could not check isClosed on connection, assuming it closed: " + e.getMessage());
276                 }
277 
278 
279 
280                 if (isClosed) {
281                     MultiConnection newCon = null;
282                     log.warn("WILL KILL SQL because connection was closed. ID=" + con.hashCode() + " SQL: " + con.lastSql);
283                     try {
284                         // get a new connection to replace this one
285                         newCon = getMultiConnection();
286                     } catch(SQLException e) {
287                         log.error("Can't add connection to pool (after close) " + e.toString());
288                     }
289                     if (newCon != null) { // successfully created new connection
290                         // we close connections in a seperate thread, for those broken JDBC drivers out there
291                         new ConnectionCloser(con);
292                     } else {
293                         // could not create new connection somewhy, but this one is **** up as well, what to do?
294                         // fail every future query:
295                         newCon = con; // simply put it back in the available pool, so everything will fail
296                     }
297                     pool.add(newCon);
298                     releaseCount++;
299                     i.remove();
300 
301                     continue;
302                 }
303 
304                 long diff = nowTime - con.getStartTimeMillis();
305 
306                 if (log.isDebugEnabled()) {
307                     if (diff > 5000 || diff > maxZeroTime) {  // don't log too often
308                         log.debug("Checking a busy connection " + con + " time = " + diff + " seconds");
309                     }
310                 }
311 
312                 if (diff < maxZeroTime) {
313                     // ok, just wait
314                 } else if (diff < maxLifeTime) {
315                     // between 30 and 120 we putback 'zero' connections
316                     if (con.lastSql == null || con.lastSql.length() == 0) {
317                         log.warn("null connection putBack " + Logging.stackTrace());
318                         pool.add(con);
319                         releaseCount++;
320                         i.remove();
321                     }
322                 } else {
323                     // above 120 we close the connection and open a new one
324                     MultiConnection newCon = null;
325                     log.warn("WILL KILL SQL. It took already " + (diff / 1000) + " seconds, which is too long. ID=" + con.hashCode() + " SQL: " + con.lastSql);
326                     try {
327                         // get a new connection to replace this one
328                         newCon = getMultiConnection();
329                     } catch(SQLException e) {
330                         log.error("Can't add connection to pool (after kill) " + e.toString());
331                     }
332                     if (newCon != null) { // successfully created new connection
333                         pool.add(newCon);
334                         releaseCount++;
335                         i.remove();
336                         // we close connections in a seperate thread, for those broken JDBC drivers out there
337                         new ConnectionCloser(con);
338                     } else {
339                         // could not create new connection somewhy, will be retried in the next cycle
340                     }
341                 }
342             }
343 
344             if ((busyPool.size() + pool.size()) != conMax) {
345                 // cannot happen, I hope...
346                 log.error("Number of connections is not correct: " + busyPool.size() + " + " + pool.size () + " = " + (busyPool.size() + pool.size()) + " != " + conMax);
347                 // Check if there are dups in the pools
348                 for(Iterator i = busyPool.iterator(); i.hasNext();) {
349                     MultiConnection bcon = (MultiConnection) i.next();
350                     int j = pool.indexOf(bcon);
351                     if (j >= 0) {
352                         if (log.isDebugEnabled()) {
353                             log.debug("duplicate connection found at " + j);
354                         }
355                         pool.remove(j);
356                     }
357                 }
358 
359                 while(((busyPool.size() + pool.size()) > conMax) && pool.size()>2) {
360                     // Remove too much ones.
361                     MultiConnection con = (MultiConnection) pool.remove(0);
362                     if (log.isDebugEnabled()) {
363                         log.debug("removing connection "+con);
364                     }
365                 }
366 
367             }
368             semaphore.release(releaseCount);
369         } // synchronized(semaphore)
370         if (log.isDebugEnabled()){
371             log.debug("finished  checkTime()");
372         }
373     }
374 
375     /**
376      * Get a free connection from the pool
377      */
378     MultiConnection getFree() {
379 
380         MultiConnection con = null;
381         try {
382             if (semaphore == null) return null; // during start-up this could happen.
383             //see comment in method checkTime()
384             synchronized (semaphore) {
385                 totalConnections++;
386                 if (conMax == 0) { // could happen during shut down of MMBase
387                     try {
388                         con = getMultiConnection(); // hm....
389                         con.claim();
390                         return con;
391                     } catch (SQLException sqe) {
392                         return null; // will probably cause NPE's but well
393                     }
394                 }
395 
396                 semaphore.acquire(); // locks until something free
397                 if (log.isDebugEnabled()) {
398                     log.debug("Getting free connection from pool " + pool.size());
399                 }
400                 con = (MultiConnection) pool.remove(0);
401                 con.claim();
402                 try {
403                     if (con.isClosed()) {
404                         con = getMultiConnection();
405                         con.claim();
406                         log.service("Got a closed connection from the connection Pool, it was replaced by a new one.");
407                     }
408                 } catch (SQLException sqe) {
409                     log.error("Closed connection could not be replaced because: " + sqe.getClass().getName() + ": " + sqe.getMessage());
410                 }
411                 busyPool.add(con);
412                 if (log.isDebugEnabled()) {
413                     log.debug("Pool: " + pool.size() + " busypool: " + busyPool.size());
414                 }
415             }
416         } catch (InterruptedException   e) {
417             log.error("GetFree was Interrupted");
418         }
419         return con;
420     }
421 
422     /**
423      * putback the used connection in the pool
424      */
425     void putBack(MultiConnection con) {
426         //see comment in method checkTime()
427         synchronized (semaphore) {
428             if (log.isDebugEnabled()) {
429                 log.debug("Putting back to Pool: " + pool.size() + " from busypool: " + busyPool.size());
430             }
431 
432             if (! busyPool.contains(con)) {
433                 // try to guess wat happend, to report that in the logs
434                 MMBase mmb = MMBase.getMMBase();
435                 if (! mmb.isShutdown()) {
436                     try {
437                         if (con.isClosed()) {
438                             log.debug("Connection " + con + " as closed an not in busypool, so it was removed from busyPool by checkTime");
439                         } else {
440                             log.warn("Put back connection (" + con.lastSql + ") was not in busyPool!!");
441                         }
442                     } catch (SQLException sqe) {
443                         log.warn("Connection " + con + " not in busypool : " + sqe.getMessage());
444                     }
445                 } else {
446                     log.service("Connection " + con.lastSql + " was put back, but MMBase is shut down, so it was ignored.");
447                 }
448                 return;
449             }
450 
451 
452             con.release(); //Only resets time connection is busy.
453 
454             MultiConnection oldCon = con;
455 
456             if (doReconnect && (con.getUsage() > maxQueries)) {
457                 log.debug("Re-oppening connection");
458 
459                 boolean gotNew = false;
460                 try {
461                     con = getMultiConnection();
462                     gotNew = true;
463                 } catch(SQLException sqe) {
464                     log.error("Can't add connection to pool (during reconnect) " + sqe.toString());
465                 }
466 
467                 if (gotNew) { // a new conection has successfully created, the old one can be closed
468                     new ConnectionCloser(oldCon);
469                 } else { // use the old one another cycle
470                     log.service("Will continue use original connection");
471                     con.resetUsage();
472                 }
473             }
474 
475             pool.add(con);
476             busyPool.remove(oldCon);
477             if (log.isDebugEnabled()) {
478                 log.debug("Pool: " + pool.size() + " busypool: " + busyPool.size());
479             }
480             semaphore.release();
481         }
482     }
483 
484     /**
485      * get the pool size
486      */
487     public int getSize() {
488         return pool.size();
489     }
490 
491     /**
492      * get the number of statements performed
493      */
494     public int getTotalConnectionsCreated() {
495         return totalConnections;
496     }
497 
498     /**
499      * For reporting purposes the connections in pool can be listed.
500      * An Iterator on a copy of the Pool is returned.
501      *
502      * @see JDBC#listConnections
503      */
504     public Iterator getPool() {
505         synchronized(semaphore) {
506             return new ArrayList(pool).iterator();
507         }
508     }
509 
510 
511     /**
512      * For reporting purposes the connections in busypool can be listed.
513      * An Iterator on a copy of the BusyPool is returned.
514      * @see JDBC#listConnections
515      */
516 
517     public Iterator getBusyPool() {
518         synchronized(semaphore) {
519             return new ArrayList(busyPool).iterator();
520         }
521     }
522 
523     /**
524      * @javadoc
525      */
526     public String   toString() {
527         return "dbm=" + url + ",name=" + name + ",conmax=" + conMax;
528     }
529 
530 
531     /**
532      * Support class to close connections in a seperate thread, as some JDBC drivers
533      * have a tendency to hang themselves on a running sql close.
534      */
535     static class ConnectionCloser implements Runnable   {
536         private static final Logger log = Logging.getLoggerInstance(ConnectionCloser.class);
537 
538         private MultiConnection connection;
539 
540         public ConnectionCloser(MultiConnection con) {
541             connection = con;
542             start();
543         }
544 
545         /**
546          * Starts a Thread and runs this Runnable
547          */
548         protected void start() {
549             // Start up the thread
550             Thread   kicker = new Thread  (this, "ConnectionCloser");
551             kicker.setDaemon(true); // For the case if indeed the close 'hangs' the thread.
552             kicker.start();
553         }
554 
555         /**
556          * Close the database connection.
557          */
558         public void run() {
559             try {
560                 connection.realclose();
561             } catch(Exception   re) {
562                 log.error("Can't close connection with ID " + connection.hashCode() + ", cause: " + re);
563             }
564             log.debug("Closed connection with ID " + connection.hashCode());
565         }
566     }
567 }
568
A to Z: JavaDoc & Examples Daily Java News & Articles Open Source Projects Open Source Codes Free Computer Books Remove Frame
Popular Tags