PoolManager


1   /*****************************************************************
2    *   Licensed to the Apache Software Foundation (ASF) under one
3    *  or more contributor license agreements.  See the NOTICE file
4    *  distributed with this work for additional information
5    *  regarding copyright ownership.  The ASF licenses this file
6    *  to you under the Apache License, Version 2.0 (the
7    *  "License"); you may not use this file except in compliance
8    *  with the License.  You may obtain a copy of the License at
9    *
10   *    http://www.apache.org/licenses/LICENSE-2.0
11   *
12   *  Unless required by applicable law or agreed to in writing,
13   *  software distributed under the License is distributed on an
14   *  "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
15   *  KIND, either express or implied.  See the License for the
16   *  specific language governing permissions and limitations
17   *  under the License.
18   ****************************************************************/
19  
20  
21  package org.apache.cayenne.conn;
22  
23  import java.io.PrintWriter  ;
24  import java.sql.Connection  ;
25  import java.sql.SQLException  ;
26  import java.util.LinkedList  ;
27  import java.util.List  ;
28  import java.util.ListIterator  ;
29  
30  import javax.sql.ConnectionEvent  ;
31  import javax.sql.ConnectionEventListener  ;
32  import javax.sql.ConnectionPoolDataSource  ;
33  import javax.sql.DataSource  ;
34  import javax.sql.PooledConnection  ;
35  
36  /**
37   * PoolManager is a pooling DataSource impementation. 
38   * Internally to obtain connections PoolManager uses either a JDBC driver 
39   * or another pooling datasource.
40   * 
41   * @author Andrus Adamchik
42   */
43  public class PoolManager implements DataSource  , ConnectionEventListener   {
44  
45      /** 
46       * Defines a maximum time in milliseconds that a connection
47       * request could wait in the connection queue. After this period
48       * expires, an exception will be thrown in the calling method.
49       * In the future this parameter should be made configurable.
50       */
51      public static final int MAX_QUEUE_WAIT = 20000;
52  
53      protected ConnectionPoolDataSource   poolDataSource;
54      protected int minConnections;
55      protected int maxConnections;
56      protected String   dataSourceUrl;
57      protected String   jdbcDriver;
58      protected String   password;
59      protected String   userName;
60  
61      protected List   unusedPool;
62      protected List   usedPool;
63  
64      private PoolMaintenanceThread poolMaintenanceThread;
65  
66  
67      /** 
68       * Creates new PoolManager using org.apache.cayenne.conn.PoolDataSource
69       * for an underlying ConnectionPoolDataSource. 
70       */
71      public PoolManager(
72          String   jdbcDriver,
73          String   dataSourceUrl,
74          int minCons,
75          int maxCons,
76          String   userName,
77          String   password)
78          throws SQLException   {
79  
80          this(jdbcDriver, dataSourceUrl, minCons, maxCons, userName, password, null);
81      }
82  
83      public PoolManager(
84          String   jdbcDriver,
85          String   dataSourceUrl,
86          int minCons,
87          int maxCons,
88          String   userName,
89          String   password,
90          ConnectionEventLoggingDelegate logger)
91          throws SQLException   {
92  
93          if (logger != null) {
94              DataSourceInfo info = new DataSourceInfo();
95              info.setJdbcDriver(jdbcDriver);
96              info.setDataSourceUrl(dataSourceUrl);
97              info.setMinConnections(minCons);
98              info.setMaxConnections(maxCons);
99              info.setUserName(userName);
100             info.setPassword(password);
101             logger.logPoolCreated(info);
102         }
103 
104         this.jdbcDriver = jdbcDriver;
105         this.dataSourceUrl = dataSourceUrl;
106         DriverDataSource driverDS = new DriverDataSource(jdbcDriver, dataSourceUrl);
107         driverDS.setLogger(logger);
108         PoolDataSource poolDS = new PoolDataSource(driverDS);
109         init(poolDS, minCons, maxCons, userName, password);
110     }
111 
112     /** Creates new PoolManager with the specified policy for
113      *  connection pooling and a ConnectionPoolDataSource object.
114      *
115      *  @param poolDataSource data source for pooled connections
116      *  @param minCons Non-negative integer that specifies a minimum number of open connections
117      *  to keep in the pool at all times
118      *  @param maxCons Non-negative integer that specifies maximum number of simultaneuosly open connections
119      *
120      *  @throws SQLException if pool manager can not be created.
121      */
122     public PoolManager(
123         ConnectionPoolDataSource   poolDataSource,
124         int minCons,
125         int maxCons,
126         String   userName,
127         String   password)
128         throws SQLException   {
129         init(poolDataSource, minCons, maxCons, userName, password);
130     }
131 
132     /** Initializes pool. Normally called from constructor. */
133     protected void init(
134         ConnectionPoolDataSource   poolDataSource,
135         int minCons,
136         int maxCons,
137         String   userName,
138         String   password)
139         throws SQLException   {
140 
141         // do sanity checks...
142         if (maxConnections < 0) {
143             throw new SQLException  (
144                 "Maximum number of connections can not be negative (" + maxCons + ").");
145         }
146 
147         if (minConnections < 0) {
148             throw new SQLException  (
149                 "Minimum number of connections can not be negative (" + minCons + ").");
150         }
151 
152         if (minConnections > maxConnections) {
153             throw new SQLException  ("Minimum number of connections can not be bigger then maximum.");
154         }
155 
156         // init properties
157         this.userName = userName;
158         this.password = password;
159         this.minConnections = minCons;
160         this.maxConnections = maxCons;
161         this.poolDataSource = poolDataSource;
162 
163         // init pool... use linked lists to use the queue in the FIFO manner
164         usedPool = new LinkedList  ();
165         unusedPool = new LinkedList  ();
166         growPool(minConnections, userName, password);
167 
168         startMaintenanceThread();
169     }
170     
171     protected synchronized void startMaintenanceThread() {
172         disposeOfMaintenanceThread();
173         this.poolMaintenanceThread = new PoolMaintenanceThread(this);
174         this.poolMaintenanceThread.start();
175     }
176 
177     /** 
178      * Creates and returns new PooledConnection object, adding itself as a listener 
179      * for connection events. 
180      */
181     protected PooledConnection   newPooledConnection(String   userName, String   password)
182         throws SQLException   {
183         PooledConnection   connection =
184             (userName != null)
185                 ? poolDataSource.getPooledConnection(userName, password)
186                 : poolDataSource.getPooledConnection();
187         connection.addConnectionEventListener(this);
188         return connection;
189     }
190 
191     /** Closes all existing connections, removes them from the pool. */
192     public void dispose() throws SQLException   {
193         synchronized (this) {
194             // clean connections from the pool
195             ListIterator   unusedIterator = unusedPool.listIterator();
196             while (unusedIterator.hasNext()) {
197                 PooledConnection   con = (PooledConnection  ) unusedIterator.next();
198                 // close connection
199                 con.close();
200                 // remove connection from the list
201                 unusedIterator.remove();
202             }
203 
204             // clean used connections
205             ListIterator   usedIterator = usedPool.listIterator();
206             while (usedIterator.hasNext()) {
207                 PooledConnection   con = (PooledConnection  ) usedIterator.next();
208                 // stop listening for connection events
209                 con.removeConnectionEventListener(this);
210                 // close connection
211                 con.close();
212                 // remove connection from the list
213                 usedIterator.remove();
214             }
215         }
216 
217         disposeOfMaintenanceThread();
218     }
219     
220     protected void disposeOfMaintenanceThread() {
221         if (poolMaintenanceThread != null) {
222             this.poolMaintenanceThread.dispose();
223         }
224     }
225 
226     /**
227      * @return true if at least one more connection can be added to the pool.
228      */
229     protected synchronized boolean canGrowPool() {
230         return getPoolSize() < maxConnections;
231     }
232 
233     /** 
234      * Increases connection pool by the specified number of connections.
235      * 
236      * @return the actual number of created connections. 
237      * @throws SQLException if an error happens when creating a new connection.
238      */
239     protected synchronized int growPool(
240         int addConnections,
241         String   userName,
242         String   password)
243         throws SQLException   {
244 
245         int i = 0;
246         int startPoolSize = getPoolSize();
247         for (; i < addConnections && startPoolSize + i < maxConnections; i++) {
248             PooledConnection   newConnection = newPooledConnection(userName, password);
249             unusedPool.add(newConnection);
250         }
251 
252         return i;
253     }
254 
255     protected synchronized void shrinkPool(int closeConnections) {
256         int idleSize = unusedPool.size();
257         for (int i = 0; i < closeConnections && i < idleSize; i++) {
258             PooledConnection   con = (PooledConnection  ) unusedPool.remove(i);
259 
260             try {
261                 con.close();
262             } catch (SQLException   ex) {
263                 // ignore
264             }
265         }
266     }
267 
268     /** 
269      * Returns maximum number of connections this pool can keep.
270      * This parameter when configured allows to limit the number of simultaneously
271      * open connections.
272      */
273     public int getMaxConnections() {
274         return maxConnections;
275     }
276 
277     public void setMaxConnections(int maxConnections) {
278         this.maxConnections = maxConnections;
279     }
280 
281     /** Returns the absolute minimum number of connections allowed 
282       * in this pool at any moment in time. */
283     public int getMinConnections() {
284         return minConnections;
285     }
286 
287     public void setMinConnections(int minConnections) {
288         this.minConnections = minConnections;
289     }
290 
291     /** Returns a database URL used to initialize this pool.
292       * Will return null if the pool was initialized with ConnectionPoolDataSource. */
293     public String   getDataSourceUrl() {
294         return dataSourceUrl;
295     }
296 
297     /** Returns a name of a JDBC driver used to initialize this pool.
298       * Will return null if the pool was initialized with ConnectionPoolDataSource. */
299     public String   getJdbcDriver() {
300         return jdbcDriver;
301     }
302 
303     /** Returns a data source password used to initialize this pool. */
304     public String   getPassword() {
305         return password;
306     }
307 
308     /** Returns a data source user name used to initialize this pool. */
309     public String   getUserName() {
310         return userName;
311     }
312 
313     /**
314      * Returns current number of connections.
315      */
316     public synchronized int getPoolSize() {
317         return usedPool.size() + unusedPool.size();
318     }
319 
320     /** 
321      * Returns the number of connections obtained via this DataSource
322      * that are currently in use by the DataSource clients. 
323      */
324     public synchronized int getCurrentlyInUse() {
325         return usedPool.size();
326     }
327 
328     /** 
329      * Returns the number of connections maintained in the 
330      * pool that are currently not used by any clients and are
331      * available immediately via <code>getConnection</code> method. 
332      */
333     public synchronized int getCurrentlyUnused() {
334         return unusedPool.size();
335     }
336 
337     /** 
338      * Returns connection from the pool using internal values of user name
339      * and password. Eqivalent to calling: 
340      * 
341      * <p><code>ds.getConnection(ds.getUserName(), ds.getPassword())</code></p> 
342      */
343     public Connection   getConnection() throws SQLException   {
344         return getConnection(userName, password);
345     }
346 
347     /** Returns connection from the pool. */
348     public synchronized Connection   getConnection(String   userName, String   password)
349             throws SQLException   {
350 
351         PooledConnection   pooledConnection = uncheckPooledConnection(userName, password);
352 
353         try {
354             return uncheckConnection(pooledConnection);
355         }
356         catch (SQLException   ex) {
357 
358             try {
359                 pooledConnection.close();
360             }
361             catch (SQLException   ignored) {
362             }
363             
364             // do one reconnect attempt...
365             pooledConnection = uncheckPooledConnection(userName, password);
366             try {
367                 return uncheckConnection(pooledConnection);
368             }
369             catch (SQLException   reconnectEx) {
370                 try {
371                     pooledConnection.close();
372                 }
373                 catch (SQLException   ignored) {
374                 }
375                 
376                 throw reconnectEx;
377             }
378         }
379     }
380     
381     private Connection   uncheckConnection(PooledConnection   pooledConnection)
382             throws SQLException   {
383         Connection   c = pooledConnection.getConnection();
384 
385         // only do that on successfully unchecked connection...
386         usedPool.add(pooledConnection);
387         return c;
388     }
389     
390     private PooledConnection   uncheckPooledConnection(String   userName, String   password)
391             throws SQLException   {
392         // wait for returned connections or the maintenance thread 
393         // to bump the pool size...
394 
395         if (unusedPool.size() == 0) {
396             
397             // first try to open a new connection
398             if (canGrowPool()) {
399                 return newPooledConnection(userName, password);
400             }
401             
402             // can't open no more... will have to wait for others to return a connection
403             
404             // note that if we were woken up 
405             // before the full wait period expired, and no connections are
406             // available yet, go back to sleep. Otherwise we don't give a maintenance
407             // thread a chance to increase pool size
408             long waitTill =
409                 System.currentTimeMillis()
410                 + MAX_QUEUE_WAIT;
411 
412             do {
413                 try {
414                     wait(MAX_QUEUE_WAIT);
415                 } catch (InterruptedException   iex) {
416                     // ignoring
417                 }
418 
419             } while (unusedPool.size() == 0 && waitTill > System.currentTimeMillis());
420 
421             if (unusedPool.size() == 0) {
422                 throw new SQLException  (
423                     "Can't obtain connection. Request timed out. Total used connections: "
424                         + usedPool.size());
425             }
426         }
427 
428         // get first connection... lets cycle them in FIFO manner
429         return (PooledConnection  ) unusedPool.remove(0);
430     }
431 
432     public int getLoginTimeout() throws java.sql.SQLException   {
433         return poolDataSource.getLoginTimeout();
434     }
435 
436     public void setLoginTimeout(int seconds) throws java.sql.SQLException   {
437         poolDataSource.setLoginTimeout(seconds);
438     }
439 
440     public PrintWriter   getLogWriter() throws java.sql.SQLException   {
441         return poolDataSource.getLogWriter();
442     }
443 
444     public void setLogWriter(PrintWriter   out) throws java.sql.SQLException   {
445         poolDataSource.setLogWriter(out);
446     }
447 
448     /** 
449      * Returns closed connection to the pool. 
450      */
451     public synchronized void connectionClosed(ConnectionEvent   event) {
452         // return connection to the pool
453         PooledConnection   closedConn = (PooledConnection  ) event.getSource();
454         
455         // remove this connection from the list of connections
456         // managed by this pool...
457         int usedInd = usedPool.indexOf(closedConn);
458         if (usedInd >= 0) {
459             usedPool.remove(usedInd);
460             unusedPool.add(closedConn);
461 
462             // notify threads waiting for connections
463             notifyAll();
464         }
465         // else ....
466         // other possibility is that this is a bad connection, so just ignore its closing event,
467         // since it was unregistered in "connectionErrorOccurred"
468     }
469 
470     /** 
471      * Removes connection with an error from the pool. This method
472      * is called by PoolManager connections on connection errors
473      * to notify PoolManager that connection is in invalid state.
474      */
475     public synchronized void connectionErrorOccurred(ConnectionEvent   event) {
476         // later on we should analyze the error to see if this
477         // is fatal... right now just kill this PooledConnection
478 
479         PooledConnection   errorSrc = (PooledConnection  ) event.getSource();
480 
481         // remove this connection from the list of connections
482         // managed by this pool...
483 
484         int usedInd = usedPool.indexOf(errorSrc);
485         if (usedInd >= 0) {
486             usedPool.remove(usedInd);
487         } else {
488             int unusedInd = unusedPool.indexOf(errorSrc);
489             if (unusedInd >= 0)
490                 unusedPool.remove(unusedInd);
491         }
492 
493         // do not close connection,
494         // let the code that catches the exception handle it
495         // ....
496     }
497 
498     static class PoolMaintenanceThread extends Thread   {
499         protected boolean shouldDie;
500         protected PoolManager pool;
501 
502         PoolMaintenanceThread(PoolManager pool) {
503             super.setName("PoolManagerCleanup-" + pool.hashCode());
504             super.setDaemon(true);
505             this.pool = pool;
506         }
507 
508         public void run() {
509             // periodically wakes up to check if the pool should grow or shrink 
510             while (true) {
511 
512                 try {
513                     // don't do it too often
514                     sleep(600000);
515                 } catch (InterruptedException   iex) {
516                     // ignore...
517                 }
518 
519                 if (shouldDie) {
520                     break;
521                 }
522 
523                 synchronized (pool) {
524                     // TODO: implement a smarter algorithm for pool management... 
525                     // right now it will simply close one connection if the count is
526                     // above median and there are any idle connections.
527 
528                     int unused = pool.getCurrentlyUnused();
529                     int used = pool.getCurrentlyInUse();
530                     int total = unused + used;
531                     int median =
532                         pool.minConnections
533                             + 1
534                             + (pool.maxConnections - pool.minConnections) / 2;
535 
536                     if (unused > 0 && total > median) {
537                         pool.shrinkPool(1);
538                     }
539                 }
540             }
541         }
542 
543         /**
544          * Stops the thread.
545          */
546         public void dispose() {
547             shouldDie = true;
548         }
549     }
550 }
551
A to Z: JavaDoc & Examples Daily Java News & Articles Open Source Projects Open Source Codes Free Computer Books Remove Frame
Popular Tags