Java API By Example, From Geeks To Geeks.

1 /** 2  * Sequoia: Database clustering technology. 3  * Copyright (C) 2002-2004 French National Institute For Research In Computer 4  * Science And Control (INRIA). 5  * Copyright (C) 2005 AmicoSoft, Inc. dba Emic Networks 6  * Copyright (C) 2005-2006 Continuent, Inc. 7  * Contact: sequoia@continuent.org 8  * 9  * Licensed under the Apache License, Version 2.0 (the "License"); 10  * you may not use this file except in compliance with the License. 11  * You may obtain a copy of the License at 12  * 13  * http://www.apache.org/licenses/LICENSE-2.0 14  * 15  * Unless required by applicable law or agreed to in writing, software 16  * distributed under the License is distributed on an "AS IS" BASIS, 17  * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. 18  * See the License for the specific language governing permissions and 19  * limitations under the License. 20  * 21  * Initial developer(s): Emmanuel Cecchet. 22  * Contributor(s): Mathieu Peltier, Sara Bouchenak, Jean-Bernard van Zuylen, Guillaume Smet. 23  */ 24 25 package org.continuent.sequoia.controller.backend; 26 27 import java.io.StringReader ; 28 import java.net.ConnectException ; 29 import java.sql.Connection ; 30 import java.sql.SQLException ; 31 import java.sql.SQLWarning ; 32 import java.sql.Savepoint ; 33 import java.sql.Statement ; 34 import java.util.ArrayList ; 35 import java.util.Collection ; 36 import java.util.Date ; 37 import java.util.HashMap ; 38 import java.util.Iterator ; 39 import java.util.List ; 40 import java.util.Map ; 41 import java.util.Vector ; 42 43 import javax.management.AttributeChangeNotification ; 44 import javax.management.MalformedObjectNameException ; 45 import javax.management.Notification ; 46 import javax.management.NotificationBroadcasterSupport ; 47 48 import org.continuent.sequoia.common.exceptions.NoTransactionStartWhenDisablingException; 49 import org.continuent.sequoia.common.exceptions.UnreachableBackendException; 50 import org.continuent.sequoia.common.exceptions.VirtualDatabaseException; 51 import org.continuent.sequoia.common.i18n.Translate; 52 import org.continuent.sequoia.common.jmx.JmxConstants; 53 import org.continuent.sequoia.common.jmx.management.BackendInfo; 54 import org.continuent.sequoia.common.jmx.management.BackendState; 55 import org.continuent.sequoia.common.jmx.monitoring.backend.BackendStatistics; 56 import org.continuent.sequoia.common.jmx.notifications.SequoiaNotificationList; 57 import org.continuent.sequoia.common.log.Trace; 58 import org.continuent.sequoia.common.sql.metadata.MetadataContainer; 59 import org.continuent.sequoia.common.users.VirtualDatabaseUser; 60 import org.continuent.sequoia.common.xml.DatabasesXmlTags; 61 import org.continuent.sequoia.common.xml.XmlComponent; 62 import org.continuent.sequoia.controller.backend.rewriting.AbstractRewritingRule; 63 import org.continuent.sequoia.controller.connection.AbstractConnectionManager; 64 import org.continuent.sequoia.controller.connection.DriverManager; 65 import org.continuent.sequoia.controller.connection.FailFastPoolConnectionManager; 66 import org.continuent.sequoia.controller.connection.PooledConnection; 67 import org.continuent.sequoia.controller.connection.RandomWaitPoolConnectionManager; 68 import org.continuent.sequoia.controller.connection.SimpleConnectionManager; 69 import org.continuent.sequoia.controller.connection.VariablePoolConnectionManager; 70 import org.continuent.sequoia.controller.core.ControllerConstants; 71 import org.continuent.sequoia.controller.loadbalancer.AbstractLoadBalancer; 72 import org.continuent.sequoia.controller.loadbalancer.BackendTaskQueues; 73 import org.continuent.sequoia.controller.loadbalancer.BackendWorkerThread; 74 import org.continuent.sequoia.controller.loadbalancer.tasks.AbstractTask; 75 import org.continuent.sequoia.controller.loadbalancer.tasks.KillThreadTask; 76 import org.continuent.sequoia.controller.requests.AbstractRequest; 77 import org.continuent.sequoia.controller.requests.AbstractWriteRequest; 78 import org.continuent.sequoia.controller.requests.StoredProcedure; 79 import org.continuent.sequoia.controller.semantic.SemanticBehavior; 80 import org.continuent.sequoia.controller.sql.schema.DatabaseSchema; 81 import org.continuent.sequoia.controller.sql.schema.DatabaseTable; 82 import org.continuent.sequoia.controller.sql.schema.DynamicDatabaseSchema; 83 import org.continuent.sequoia.controller.virtualdatabase.VirtualDatabase; 84 import org.dom4j.Document; 85 import org.dom4j.Element; 86 import org.dom4j.io.SAXReader; 87 88 /** 89  * A <code>DatabaseBackend</code> represents a real database backend that will 90  * have to be bound to a virtual Sequoia database. All connections opened will 91  * use the same url but possibly different login/password. 92  * 93  * @author <a HREF="mailto:Emmanuel.Cecchet@inria.fr">Emmanuel Cecchet </a> 94  * @author <a HREF="mailto:Mathieu.Peltier@inrialpes.fr">Mathieu Peltier </a> 95  * @author <a HREF="mailto:Sara.Bouchenak@epfl.ch">Sara Bouchenak </a> 96  * @author <a HREF="mailto:Nicolas.Modrzyk@inrialpes.fr">Nicolas Modrzyk </a> 97  * @author <a HREF="mailto:jbvanzuylen@transwide.com">Jean-Bernard van Zuylen 98  * </a> 99  * @author <a HREF="mailto:guillaume.smet@gmail.com">Guillaume Smet</a> 100  * @version 1.0 101  */ 102 public final class DatabaseBackend implements XmlComponent 103 { 104   // 105 // How the code is organized? 106 // 1. Member variables 107 // 2. Constructor(s) 108 // 3. Connection management 109 // 4. Transaction management 110 // 5. State management 111 // 6. Schema management 112 // 7. Monitoring 113 // 8. Getter/Setter (possibly in alphabetical order) 114 // 9. Worker threads management 115 // 116 117   /** Logical name assigned to this backend. */ 118   private String name; 119 120   /** Path for driver */ 121   private String driverPath; 122 123   /** Database native JDBC driver class name. */ 124   private String driverClassName; 125 126   /** Driver compliance to Sequoia requirements */ 127   private transient DriverCompliance driverCompliance; 128 129   /** Real URL to access the database (JDBC URL). */ 130   private String url; 131 132   /** Virtual database this backend is attached to */ 133   private VirtualDatabase vdb; 134 135   /** A boolean to know if we should allow this backend to be enabled for write */ 136   private boolean writeCanBeEnabled; 137 138   /** SQL statement used to check if a connection is still valid */ 139   private String connectionTestStatement; 140 141   /** 142    * The schema of the database. This should be accessed in a synchronized(this) 143    * block since it can be updated dynamically. 144    */ 145   private transient DatabaseSchema schema; 146 147   /** <code>true</code> if schema is no more up-to-date and needs a refresh */ 148   private boolean schemaIsDirty = true; 149 150   /** Connection managers for this backend. */ 151   private transient Map connectionManagers; 152 153   /** List of persistent connections active for this backend (<persistentConnectionId,PooledConnection>) */ 154   private Map persistentConnections; 155 156   /** Logger instance. */ 157   protected transient Trace logger; 158 159   /** 160    * ArrayList&lt;Long&gt; of active transactions on this backend (as opposed to 161    * the transactions started on the virtual database managing this backend). 162    */ 163   private transient ArrayList activeTransactions = new ArrayList (); 164 165   /** List of savepoints for each transaction */ 166   private transient Map savepoints = new HashMap (); 167 168   /** List of pending requests. */ 169   private transient Vector pendingRequests = new Vector (); 170 171   /** List of pending tasks. */ 172   private transient Vector pendingTasks = new Vector (); 173 174   /** Task queues to handle writes in RAIDb-1 or RAIDb-2 */ 175   private transient BackendTaskQueues taskQueues = null; 176   /** List of BackendWorkerThread to execute write queries in RAIDb-1 or RAIDb-2 */ 177   private int nbOfWorkerThreads = 5; 178   private ArrayList workerThreads = null; 179   private final Object workerThreadSync = new Object (); 180 181   /** Monitoring Values */ 182   private int totalRequest; 183   private int totalWriteRequest; 184   private int totalReadRequest; 185   private int totalTransactions; 186 187   /** List of <code>AbstractRewritingRule</code> objects. */ 188   private ArrayList rewritingRules; 189 190   /** For metadata information generation */ 191   private int dynamicPrecision = -1; 192 193   /** Short form of SQL statements to include in traces and exceptions */ 194   private int sqlShortFormLength = 40; 195 196   private String lastKnownCheckpoint; 197 198   /** 199    * The current state of the backend 200    * 201    * @see org.continuent.sequoia.common.jmx.management.BackendState 202    */ 203   private int state = BackendState.DISABLED; 204 205   private transient BackendStateListener stateListener; 206 207   private NotificationBroadcasterSupport notificationBroadcaster; 208 209   private int notificationSequence = 0; 210 211   private int totalTasks; 212 213   private AbstractConnectionManager defaultConnectionManager = null; 214 215   /* end user logger */ 216   static Trace endUserLogger = Trace 217                                                                       .getLogger("org.continuent.sequoia.enduser"); 218 219   /** 220    * Creates a new <code>DatabaseBackend</code> instance. 221    * 222    * @param vdb the virtual database this backend belongs to 223    * @param name logical name assigned to this backend 224    * @param driverPath path for driver 225    * @param driverClassName class name of the database native JDBC driver to 226    * load 227    * @param url URL to access the database 228    * @param writeCanBeEnabled if writes can be enabled on this backend 229    * @param connectionTestStatement SQL statement used to check if a connection 230    * is still valid 231    * @param nbOfWorkerThreads defines the number of BackendWorkerThread that 232    * should process writes in parallel (minimum is 2) 233    */ 234   public DatabaseBackend(VirtualDatabase vdb, String name, String driverPath, 235       String driverClassName, String url, boolean writeCanBeEnabled, 236       String connectionTestStatement, int nbOfWorkerThreads) 237   { 238     if (name == null) 239       throw new IllegalArgumentException (Translate 240           .get("backend.null.backend.name")); 241 242     if (driverClassName == null) 243       throw new IllegalArgumentException (Translate.get("backend.null.driver")); 244 245     if (url == null) 246       throw new IllegalArgumentException (Translate.get("backend.null.url")); 247 248     if (vdb == null) 249       throw new IllegalArgumentException (Translate 250           .get("backend.null.virtualdatabase")); 251 252     if (connectionTestStatement == null) 253       throw new IllegalArgumentException (Translate 254           .get("backend.null.connection.test")); 255 256     logger = Trace 257         .getLogger("org.continuent.sequoia.controller.backend.DatabaseBackend." 258             + name); 259 260     if (nbOfWorkerThreads < 2) 261     { 262       nbOfWorkerThreads = 2; 263       logger.warn("Invalid number of worker threads (" + nbOfWorkerThreads 264           + "), re-adjusting to the minimum (2)."); 265     } 266 267     this.name = name; 268     this.writeCanBeEnabled = writeCanBeEnabled; 269     this.driverPath = driverPath; 270     this.driverClassName = driverClassName; 271     this.url = url; 272     this.vdb = vdb; 273     this.connectionTestStatement = connectionTestStatement; 274     this.nbOfWorkerThreads = nbOfWorkerThreads; 275     this.connectionManagers = new HashMap (); 276     this.persistentConnections = new HashMap (); 277     this.driverCompliance = new DriverCompliance(logger); 278     totalRequest = 0; 279   } 280 281   /** 282    * Creates a new <code>DatabaseBackend</code> object 283    * 284    * @param info a backend info object to create a database backend object from 285    */ 286   public DatabaseBackend(VirtualDatabase vdb, BackendInfo info) 287   { 288     this(vdb, info.getName(), info.getDriverPath(), info.getDriverClassName(), 289         info.getUrl(), true, info.getConnectionTestStatement(), info 290             .getNbOfWorkerThreads()); 291     try 292     { 293       String xml = info.getXml(); 294       StringReader sreader = new StringReader (xml); 295       SAXReader reader = new SAXReader(); 296       Document document = reader.read(sreader); 297       Element root = document.getRootElement(); 298       Iterator iter1 = root.elementIterator(); 299       while (iter1.hasNext()) 300       { 301         Element elem = (Element) iter1.next(); 302         if (elem.getName().equals(DatabasesXmlTags.ELT_ConnectionManager)) 303         { 304           String vuser = elem.valueOf("@" + DatabasesXmlTags.ATT_vLogin); 305           String rlogin = elem.valueOf("@" + DatabasesXmlTags.ATT_rLogin); 306           String rpassword = elem.valueOf("@" + DatabasesXmlTags.ATT_rPassword); 307           Iterator iter2 = elem.elementIterator(); 308           while (iter2.hasNext()) 309           { 310             Element connectionManager = (Element) iter2.next(); 311             String cname = connectionManager.getName(); 312             if (cname 313                 .equals(DatabasesXmlTags.ELT_VariablePoolConnectionManager)) 314             { 315               int minPoolSize = Integer.parseInt(connectionManager.valueOf("@" 316                   + DatabasesXmlTags.ATT_minPoolSize)); 317               int maxPoolSize = Integer.parseInt(connectionManager.valueOf("@" 318                   + DatabasesXmlTags.ATT_maxPoolSize)); 319               int idleTimeout = Integer.parseInt(connectionManager.valueOf("@" 320                   + DatabasesXmlTags.ATT_idleTimeout)); 321               int waitTimeout = Integer.parseInt(connectionManager.valueOf("@" 322                   + DatabasesXmlTags.ATT_waitTimeout)); 323               this.addConnectionManager(vuser, 324                   new VariablePoolConnectionManager(url, name, rlogin, 325                       rpassword, driverPath, driverClassName, minPoolSize, 326                       maxPoolSize, idleTimeout, waitTimeout)); 327             } 328             else if (cname.equals(DatabasesXmlTags.ELT_SimpleConnectionManager)) 329             { 330               this.addConnectionManager(vuser, new SimpleConnectionManager(url, 331                   name, rlogin, rpassword, driverPath, driverClassName)); 332             } 333             else if (cname 334                 .equals(DatabasesXmlTags.ELT_RandomWaitPoolConnectionManager)) 335             { 336               int poolSize = Integer.parseInt(connectionManager.valueOf("@" 337                   + DatabasesXmlTags.ATT_poolSize)); 338               int timeout = Integer.parseInt(connectionManager.valueOf("@" 339                   + DatabasesXmlTags.ATT_timeout)); 340               this 341                   .addConnectionManager(vuser, 342                       new RandomWaitPoolConnectionManager(url, name, rlogin, 343                           rpassword, driverPath, driverClassName, poolSize, 344                           timeout)); /* Lookup functions */ 345 346             } 347             else if (cname 348                 .equals(DatabasesXmlTags.ELT_FailFastPoolConnectionManager)) 349             { 350               int poolSize = Integer.parseInt(connectionManager.valueOf("@" 351                   + DatabasesXmlTags.ATT_poolSize)); 352               this.addConnectionManager(vuser, 353                   new FailFastPoolConnectionManager(url, name, rlogin, 354                       rpassword, driverPath, driverClassName, poolSize)); 355             } 356           } 357         } 358       } 359 360     } 361     catch (Exception e) 362     { 363       logger 364           .error(Translate.get("backend.add.connection.manager.failed", e), e); 365     } 366   } 367 368   /** 369    * Returns a deeply copied clone of this backend. Will use the same rewriting 370    * rules and will get new instance of connection managers with the same 371    * configuration 372    * 373    * @param newName the new name for this new backend 374    * @param parameters a set of parameters to use to replace values from the 375    * copied backend. <br> 376    * The different parameters are: <br> 377    * <ul> 378    * <li><tt>driverPath</tt>: the path to the driver</li> 379    * <li><tt>driver</tt>: the driver class name</li> 380    * <li><tt>url</tt>: the url to connect to the database</li> 381    * <li><tt>connectionTestStatement</tt>: the query to test the 382    * connection</li> 383    * </ul> 384    * <br> 385    * @return <code>DatabaseBackend</code> instance 386    * @throws Exception if cannot proceed the copy 387    */ 388   public DatabaseBackend copy(String newName, Map parameters) throws Exception 389   { 390     // Get the parameters from the backend if they are not specified, or take 391 // them from the map of parameters otherwise. 392 String fromDriverPath = parameters 393         .containsKey(DatabasesXmlTags.ATT_driverPath) ? (String ) parameters 394         .get(DatabasesXmlTags.ATT_driverPath) : this.getDriverPath(); 395 396     String fromDriverClassName = parameters 397         .containsKey(DatabasesXmlTags.ATT_driver) ? (String ) parameters 398         .get(DatabasesXmlTags.ATT_driver) : this.getDriverClassName(); 399 400     String fromUrl = parameters.containsKey(DatabasesXmlTags.ATT_url) 401         ? (String ) parameters.get(DatabasesXmlTags.ATT_url) /* Lookup functions */ 402 403         : this.getURL(); 404 405     String fromConnectionTestStatement = parameters 406         .containsKey(DatabasesXmlTags.ATT_connectionTestStatement) 407         ? (String ) parameters.get(DatabasesXmlTags.ATT_connectionTestStatement) 408         : this.getConnectionTestStatement(); 409 410     if (getURL().equals(fromUrl) 411         && getDriverClassName().equals(fromDriverClassName)) 412       throw new VirtualDatabaseException( 413           "It is not allowed to clone a backend with the same URL and driver class name"); 414 415     // Create the new backend object 416 DatabaseBackend newBackend = new DatabaseBackend(vdb, newName, 417         fromDriverPath, fromDriverClassName, fromUrl, writeCanBeEnabled, 418         fromConnectionTestStatement, nbOfWorkerThreads); 419 420     // Set the rewriting rules and the connection managers as the backend we 421 // are copying 422 newBackend.rewritingRules = this.rewritingRules; 423 424     // Set Connection managers 425 Map fromConnectionManagers = this.connectionManagers; 426     Iterator iter = fromConnectionManagers.keySet().iterator(); 427 428     String vlogin = null; 429     AbstractConnectionManager connectionManager; 430     while (iter.hasNext()) 431     { 432       vlogin = (String ) iter.next(); 433       connectionManager = (AbstractConnectionManager) fromConnectionManagers 434           .get(vlogin); 435       newBackend.addConnectionManager(vlogin, connectionManager.copy(fromUrl, 436           newName)); 437     } 438 439     return newBackend; 440   } 441 442   /** 443    * Two database backends are considered equal if they have the same name, URL 444    * and driver class name. 445    * 446    * @param other an object 447    * @return a <code>boolean</code> value 448    */ 449   public boolean equals(Object other) 450   { 451     if ((other == null) || (!(other instanceof DatabaseBackend))) 452       return false; 453     else 454     { 455       DatabaseBackend b = (DatabaseBackend) other; 456       return name.equals(b.getName()) 457           && driverClassName.equals(b.getDriverClassName()) 458           && url.equals(b.getURL()); 459     } 460   } 461 462   /** 463    * @see java.lang.Object#hashCode() 464    */ 465   public int hashCode() 466   { 467     return name.hashCode(); 468   } 469 470   // 471 // Connection management 472 // 473 474   /** 475    * Adds a <code>ConnectionManager</code> to this backend. Note that the 476    * <code>ConnectionManager</code> is not initialized in this method. 477    * 478    * @param vLogin the virtual login corresponding to this connection manager 479    * @param connectionManager the <code>ConnectionManager</code> to add 480    */ 481   public void addConnectionManager(String vLogin, 482       AbstractConnectionManager connectionManager) 483   { 484     if (connectionManager == null) 485       throw new IllegalArgumentException (Translate.get( 486           "backend.null.connection.manager", new String []{name, url})); 487     if (logger.isInfoEnabled()) 488       logger.info(Translate.get("backend.add.connection.manager.for.user", 489           vLogin)); 490     connectionManager.setVLogin(vLogin); 491     connectionManager.setConnectionTestStatement(getConnectionTestStatement()); 492     connectionManagers.put(vLogin, connectionManager); 493   } 494 495   /** 496    * Adds a default connection manager on this backend for the specified user. 497    * The default connection manager should be specified in the vdb config file. 498    * 499    * @param vdbUser user for whom the connnection manager will be added. 500    * @throws SQLException if connection manager could not be initialized. 501    */ 502   public void addDefaultConnectionManager(VirtualDatabaseUser vdbUser) 503       throws SQLException 504   { 505     if (defaultConnectionManager == null) 506     { 507       if (logger.isWarnEnabled()) 508       { 509         logger 510             .warn("Default connection manager undefined in backend configuration, setting to a VariablePoolConnectionManager"); 511       } 512       defaultConnectionManager = new VariablePoolConnectionManager(url, name, 513           vdbUser.getLogin(), vdbUser.getPassword(), driverPath, 514           driverClassName, 20, 5, 0, 180, 0); 515     } 516     AbstractConnectionManager connectionManager = defaultConnectionManager 517         .clone(vdbUser.getLogin(), vdbUser.getPassword()); 518     connectionManager.initializeConnections(); 519     addConnectionManager(vdbUser.getLogin(), connectionManager); 520   } 521 522   /** 523    * Removes a <code>ConnectionManager</code> to this backend. Note that the 524    * <code>ConnectionManager</code> is finalized in this method. 525    * 526    * @param vdbUser the virtual user corresponding to this connection manager 527    * @throws SQLException if connection manager could not be finalized. 528    */ 529   public void removeConnectionManager(VirtualDatabaseUser vdbUser) 530       throws SQLException 531   { 532     ((AbstractConnectionManager) connectionManagers.get(vdbUser)) 533         .finalizeConnections(); 534     connectionManagers.remove(vdbUser); 535   } 536 537   /** 538    * Check if the driver used by this backend is compliant with Sequoia needs. 539    * 540    * @throws SQLException if the driver is not compliant 541    */ 542   public void checkDriverCompliance() throws SQLException 543   { 544     if (connectionManagers.isEmpty()) 545       throw new SQLException (Translate.get("backend.null.connection.manager", 546           new String []{name, url})); 547 548     AbstractConnectionManager connectionManager; 549     Iterator iter = connectionManagers.values().iterator(); 550     connectionManager = (AbstractConnectionManager) iter.next(); 551 552     try 553     { 554       if (!driverCompliance.complianceTest(url, connectionManager.getLogin(), 555           connectionManager.getPassword(), connectionManager.getDriverPath(), 556           connectionManager.getDriverClassName(), connectionTestStatement)) 557         throw new SQLException (Translate.get("backend.driver.not.compliant", 558             driverClassName, ControllerConstants.PRODUCT_NAME)); 559     } 560     catch (ConnectException e) 561     { 562       throw (SQLException ) new SQLException (Translate.get( 563           "backend.cannot.connect.to", e)).initCause(e); 564     } 565   } 566 567   /** 568    * Sets the default connection manager used for the transparent login feature. 569    * 570    * @param defaultConnectionManager connection manager 571    */ 572   public void setDefaultConnectionManager( 573       AbstractConnectionManager defaultConnectionManager) 574   { 575     this.defaultConnectionManager = defaultConnectionManager; 576   } 577 578   /** 579    * Initializes the connection managers' connections. The caller must ensure 580    * that the driver has already been loaded else an exception will be thrown. 581    * 582    * @exception SQLException if an error occurs 583    */ 584   public synchronized void initializeConnections() throws SQLException 585   { 586     if (connectionManagers.isEmpty()) 587       throw new SQLException (Translate.get("backend.not.defined", new String []{ 588           name, url})); 589 590     AbstractConnectionManager connectionManager; 591     Iterator iter = connectionManagers.values().iterator(); 592     while (iter.hasNext()) 593     { 594       connectionManager = (AbstractConnectionManager) iter.next(); 595       if (!connectionManager.isInitialized()) 596         connectionManager.initializeConnections(); 597     } 598   } 599 600   /** 601    * Releases all the connections to the database held by the connection 602    * managers. 603    * 604    * @throws SQLException if an error occurs 605    */ 606   public synchronized void finalizeConnections() throws SQLException 607   { 608     /* 609      * Remove our references to persistentConnections so that they can garbage 610      * collected. 611      */ 612     synchronized (persistentConnections) 613     { 614       persistentConnections.clear(); 615     } 616     if (connectionManagers.isEmpty()) 617       throw new SQLException (Translate.get("backend.not.defined", new String []{ 618           name, url})); 619 620     AbstractConnectionManager connectionManager; 621     Iterator iter = connectionManagers.values().iterator(); 622     while (iter.hasNext()) 623     { 624       connectionManager = (AbstractConnectionManager) iter.next(); 625       if (connectionManager.isInitialized()) 626         connectionManager.finalizeConnections(); 627     } 628   } 629 630   /** 631    * Force all connections to be renewed when they are used next. This just sets 632    * a flag and connections will be lazily replaced as needed. 633    */ 634   public void flagAllConnectionsForRenewal() 635   { 636     AbstractConnectionManager connectionManager; 637     Iterator iter = connectionManagers.values().iterator(); 638     while (iter.hasNext()) 639     { 640       connectionManager = (AbstractConnectionManager) iter.next(); 641       if (connectionManager.isInitialized()) 642         connectionManager.flagAllConnectionsForRenewal(); 643     } 644   } 645 646   /** 647    * mutex used only as a synchronization point for getting a connection in 648    * getConnection...IfNeeded() method.<br /> 649    * This method uses a synchronized block on this mutex rather than 650    * synchronizing on <code>this</code> so that we won't hang all the system 651    * if the method blocks because no connections are available. 652    * 653    * @see #getConnectionForTransactionAndLazyBeginIfNeeded(AbstractRequest, 654    * AbstractConnectionManager) 655    */ 656   private final Object connectionMutex = new Object (); 657 658   /** 659    * Add a persistent connection to this backend 660    * 661    * @param persistentConnectionId id of the persistent connection to add 662    * @param c the persistent connection to add 663    */ 664   public void addPersistentConnection(long persistentConnectionId, 665       PooledConnection c) 666   { 667     synchronized (persistentConnections) 668     { 669       persistentConnections.put(new Long (persistentConnectionId), c); 670     } 671   } 672 673   /** 674    * Returns <code>true</code> if this DatabaseBackend has persistent 675    * connections, <code>false</code> else. 676    * 677    * @return <code>true</code> if this DatabaseBackend has persistent 678    * connections, <code>false</code> else 679    */ 680   public boolean hasPersistentConnections() 681   { 682     return (persistentConnections.size() > 0); 683   } 684 685   /** 686    * Remove a persistent connection from this backend 687    * 688    * @param persistentConnectionId id of the persistent connection to remove 689    */ 690   public void removePersistentConnection(long persistentConnectionId) 691   { 692     synchronized (persistentConnections) 693     { 694       persistentConnections.remove(new Long (persistentConnectionId)); 695       if (persistentConnections.isEmpty()) 696       { 697         persistentConnections.notifyAll(); 698       } 699     } 700   } 701 702   /** 703    * Retrieve a connection for a given transaction or create a new connection 704    * and start a new transaction. <br> 705    * This method is internally synchronized on a mutex so that concurrent writes 706    * within the same transaction that are allowed to execute out of order will 707    * not open separate connection if they race on transaction begin. 708    * 709    * @param request request that will execute (must carry transaction id and 710    * transaction isolation level (does nothing if equals to 711    * Connection.DEFAULT_TRANSACTION_ISOLATION_LEVEL)) 712    * @param cm connection manager to get the connection from 713    * @return the connection for the given transaction id 714    * @throws UnreachableBackendException if the backend is no more reachable 715    * @throws NoTransactionStartWhenDisablingException if a new transaction 716    * needed to be started but the backend is in the disabling state 717    * @throws SQLException if another error occurs 718    * @see #connectionMutex 719    */ 720   public Connection getConnectionForTransactionAndLazyBeginIfNeeded( 721       AbstractRequest request, AbstractConnectionManager cm) 722       throws UnreachableBackendException, 723       NoTransactionStartWhenDisablingException, SQLException 724   { 725     Long tid = new Long (request.getTransactionId()); 726     /* 727      * An existing transaction that already has a connection must not be blocked 728      * by transactions that are waiting for connections. Transactions that have 729      * started and have a connection are past the point where concurrent 730      * requests are a problem. 731      */ 732     if (isStartedTransaction(tid)) 733     { 734       PooledConnection pc = cm.retrieveConnectionForTransaction(request 735           .getTransactionId()); 736       if (pc != null) 737       { 738         return pc.getConnection(); 739       } 740     } 741     synchronized (connectionMutex) 742     { 743       /* 744        * Repeat the test from above, because a concurrent write request may have 745        * changed the transaction status while this request was waiting for the 746        * mutex. 747        */ 748       if (isStartedTransaction(tid)) 749       { // Transaction has already been started, retrieve connection 750 PooledConnection pc = cm.retrieveConnectionForTransaction(request 751             .getTransactionId()); 752         if ((pc == null) && isDisabling()) 753         { 754           /** 755            * The backend is disabling and this transaction was added at 756            * disabling time because it was already logged but we didn't lazily 757            * start it yet, so let's do the lazy start now. 758            * 759            * @see RequestManager#disableBackendStoreCheckpointAndSetDisabling(DatabaseBackend, 760            * String) 761            */ 762           Connection c = AbstractLoadBalancer.getConnectionAndBeginTransaction( 763               this, cm, request); 764           if (c == null) 765           { 766             if (logger.isWarnEnabled()) 767             { 768               logger 769                   .warn("Null connection returned from AbstractLoadBalancer.getConnectionAndBeginTransaction() (backend is disabling)"); 770             } 771           } 772           return c; 773         } 774         if ((pc == null) && !isDisabling()) 775         { // Sanity check, should never happen 776 if (logger.isErrorEnabled()) 777           { 778             logger.error("Null connection [tid = " + request.getTransactionId() 779                 + ", backend state = " + state + ", active tx = " 780                 + activeTransactions + "]"); 781           } 782           return null; 783         } 784         return pc.getConnection(); 785       } 786       else 787       { 788         if (!canAcceptTasks(request)) 789           throw new NoTransactionStartWhenDisablingException(); 790 791         Connection c = null; 792 793         // Transaction has not been started yet, this is a lazy begin 794 c = AbstractLoadBalancer.getConnectionAndBeginTransaction(this, cm, 795             request); 796         // begin transaction 797 startTransaction(tid); 798 799         if (c == null) 800         { 801           if (logger.isWarnEnabled()) 802           { 803             logger 804                 .warn("Null connection returned from AbstractLoadBalancer.getConnectionAndBeginTransaction() [state = " 805                     + state + "]"); 806           } 807         } 808         return c; 809       } 810     } // end of synchronized (connectionMutex) 811 } 812 813   /** 814    * Returns the <code>ConnectionManager</code> associated to this backend for 815    * a given virtual login. 816    * 817    * @param vLogin the virtual login 818    * @return an <code>AbstractConnectionManager</code> instance 819    */ 820   public AbstractConnectionManager getConnectionManager(String vLogin) 821   { 822     return (AbstractConnectionManager) connectionManagers.get(vLogin); 823   } 824 825   /** 826    * Returns the SQL statement to use to check the connection validity. 827    * 828    * @return a <code>String</code> containing a SQL statement 829    */ 830   public String getConnectionTestStatement() 831   { 832     return connectionTestStatement; 833   } 834 835   /** 836    * Retrieves the SQLWarning chained to the given persistent connection 837    * 838    * @param connId persistent connection id to retrieve warnings from 839    * @exception SQLException if a database access error occurs or this method is 840    * called on a closed connection 841    * @return the warnings on the given connection or null 842    */ 843   public SQLWarning getPersistentConnectionWarnings(long connId) 844       throws SQLException 845   { 846     if (persistentConnections != null) 847     { 848       PooledConnection pc = (PooledConnection) persistentConnections 849           .get(new Long (connId)); 850       if (pc != null) 851         return pc.getConnection().getWarnings(); 852     } 853     return null; 854   } 855 856   /** 857    * Clears the SQLWarning chained to the given persistent connection 858    * 859    * @param connId persistent connection id to retrieve warnings from 860    * @exception SQLException if a database access error occurs 861    */ 862   public void clearPersistentConnectionWarnings(long connId) 863       throws SQLException 864   { 865     if (persistentConnections != null) 866     { 867       PooledConnection pc = (PooledConnection) persistentConnections 868           .get(new Long (connId)); 869       if (pc != null) 870         pc.getConnection().clearWarnings(); 871     } 872   } 873 874   /** 875    * Check if the given connection is valid or not. This function issues the 876    * connectionTestStatement query on the connection and if it succeeds then the 877    * connection is declared valid. If an exception occurs, the connection is 878    * declared invalid. 879    * 880    * @param connection the connection to test 881    * @return true if the connection is valid 882    */ 883   public boolean isValidConnection(Connection connection) 884   { 885     try 886     { 887       Statement s = connection.createStatement(); 888       s.executeQuery(connectionTestStatement); 889     } 890     catch (SQLException e) 891     { 892       if ("25P02".equals(e.getSQLState()) 893           || (e.getMessage() != null && e.getMessage().indexOf( 894               "ignored until end of transaction block") > 0)) 895       { 896         // see bug item #300873 on the forge for details 897 // PostgreSQL throws an exception if a query is issued after a request 898 // has failed within a transaction, we now have to check for this 899 // exception as it means the connection is valid 900 // 901 // PostgreSQL versions after 7.4 will return the SQLState if version 3 902 // of the protocol is used (it is the default version unless you use 903 // protocolVersion parameter in your JDBC url), whereas PostgreSQL 904 // versions prior to 7.4 (version 2 of the protocol) will have to be 905 // checked for the message text 906 return true; 907       } 908       return false; 909     } 910     return true; 911   } 912 913   /** 914    * Tell all the connection managers to stop handing out new Connections. 915    */ 916   public synchronized void shutdownConnectionManagers() 917   { 918     for (Iterator iter = connectionManagers.values().iterator(); iter.hasNext();) 919     { 920       AbstractConnectionManager cm = (AbstractConnectionManager) iter.next(); 921       if (cm.isInitialized()) 922       { 923         cm.shutdown(); 924       } 925     } 926   } 927 928   // 929 // Transaction management 930 // 931 932   /** 933    * Adds a savepoint to a given transaction 934    * 935    * @param tid transaction identifier 936    * @param savepoint savepoint to add 937    */ 938   public void addSavepoint(Long tid, Savepoint savepoint) 939   { 940     synchronized (savepoints) 941     { 942       List savepointList = (List ) savepoints.get(tid); 943       if (savepointList == null) 944       { // Lazy list creation 945 savepointList = new ArrayList (); 946         savepoints.put(tid, savepointList); 947       } 948       savepointList.add(savepoint); 949     } 950   } 951 952   /** 953    * Returns a Savepoint identified its name for a given transaction or 954    * <code>null</code> if no such Savepoint exists. 955    * 956    * @param tid transaction identifier 957    * @param savepointName name of the savepoint 958    * @return a savepoint or <code>null</code> if there is no savepoint of the 959    * given <code>savePointName</code> for the transaction identified 960    * by the <code>tid</code> 961    */ 962   public Savepoint getSavepoint(Long tid, String savepointName) 963   { 964     synchronized (savepoints) 965     { 966       List savepointList = (List ) savepoints.get(tid); 967       if (savepointList == null) 968         return null; // No checkpoint for that transaction 969 970       Iterator i = savepointList.iterator(); 971       while (i.hasNext()) 972       { 973         try 974         { 975           Savepoint savepoint = (Savepoint ) i.next(); 976           if (savepointName.equals(savepoint.getSavepointName())) 977             return savepoint; 978         } 979         catch (SQLException ignore) 980         { 981           // We should never get here because we always use named savepoints 982 // on backends 983 } 984       } 985     } 986 987     // No savepoint has been found for given savepoint name 988 return null; 989   } 990 991   /** 992    * Returns <code>true</code> if the specified transaction has been started 993    * on this backend (a connection has been allocated for this transaction). 994    * 995    * @param tid transaction identifier 996    * @return <code>true</code> if the transaction has been started 997    */ 998   public boolean isStartedTransaction(Long tid) 999   { 1000    synchronized (activeTransactions) 1001    { 1002      return activeTransactions.contains(tid); 1003    } 1004  } 1005 1006  /** 1007   * Removes a savepoint for a given transaction 1008   * 1009   * @param tid transaction identifier 1010   * @param savepoint savepoint to remove 1011   */ 1012  public void removeSavepoint(Long tid, Savepoint savepoint) 1013  { 1014    synchronized (savepoints) 1015    { 1016      List savepointList = (List ) savepoints.get(tid); 1017      if (savepointList == null) 1018        logger.error("No savepoints found for transaction " + tid); 1019      else 1020        savepointList.remove(savepoint); 1021    } 1022  } 1023 1024  /** 1025   * Signals that a transaction has been started on this backend. It means that 1026   * a connection has been allocated for this transaction. 1027   * 1028   * @param tid transaction identifier 1029   */ 1030  public void startTransaction(Long tid) 1031  { 1032    synchronized (activeTransactions) 1033    { 1034      totalTransactions++; 1035      activeTransactions.add(tid); 1036    } 1037  } 1038 1039  /** 1040   * Signals that a transaction has been stopped on this backend. It means that 1041   * the connection has been released for this transaction. 1042   * 1043   * @param tid transaction identifier 1044   */ 1045  public void stopTransaction(Long tid) 1046  { 1047    synchronized (activeTransactions) 1048    { 1049      if (!activeTransactions.remove(tid)) 1050        throw new IllegalArgumentException (Translate.get( 1051            "backend.transaction.not.started", new String []{"" + tid, name})); 1052      // If this was the last open transaction, we notify people possibly 1053 // waiting on waitForAllTransactionsToComplete() 1054 if (activeTransactions.isEmpty()) 1055      { 1056        activeTransactions.notifyAll(); 1057      } 1058    } 1059 1060    synchronized (savepoints) 1061    { 1062      savepoints.remove(tid); 1063    } 1064  } 1065 1066  /** 1067   * This method waits until all currently open transactions and persistent 1068   * connections on this backend complete. If no transaction are currently 1069   * running on this backend or no persistent connection is open, this method 1070   * immediately returns. 1071   */ 1072  public void waitForAllTransactionsAndPersistentConnectionsToComplete() 1073  { 1074    // Wait for active transactions to complete 1075 synchronized (activeTransactions) 1076    { 1077      if (!activeTransactions.isEmpty()) 1078      { 1079        if (logger.isInfoEnabled()) 1080          logger.info("Backend " + name + " wait for " 1081              + activeTransactions.size() 1082              + " transactions to complete before disabling completely."); 1083 1084        try 1085        { 1086          activeTransactions.wait(); 1087        } 1088        catch (InterruptedException ignore) 1089        { 1090        } 1091      } 1092    } 1093 1094    // Wait for active persistent connections to close 1095 synchronized (persistentConnections) 1096    { 1097      if (!persistentConnections.isEmpty()) 1098      { 1099        if (logger.isInfoEnabled()) 1100          logger 1101              .info("Backend " 1102                  + name 1103                  + " wait for " 1104                  + persistentConnections.size() 1105                  + " persistent connections to close before disabling completely."); 1106 1107        try 1108        { 1109          persistentConnections.wait(); 1110        } 1111        catch (InterruptedException ignore) 1112        { 1113        } 1114      } 1115    } 1116  } 1117 1118  // 1119 // State Management 1120 // 1121 1122  /** 1123   * Returns true if the backend is in a state that allows it to accept tasks or 1124   * the execution of the specified request. 1125   * 1126   * @param request the request that is going to execute in that task (null if 1127   * not applicable). 1128   * @return Returns true if backend isReadEnabled() or isWriteEnabled() or 1129   * isReplaying() 1130   */ 1131  public boolean canAcceptTasks(AbstractRequest request) 1132  { 1133    if (request != null) 1134      return canAcceptTasks(request.isPersistentConnection(), request 1135          .getPersistentConnectionId()); 1136    else 1137      return canAcceptTasks(false, -1); 1138  } 1139 1140  /** 1141   * Returns true if the backend is in a state that allows it to accept tasks or 1142   * the execution for the specified persistent connection. 1143   * 1144   * @param persistentConnectionId the persistent connection id of the request 1145   * that is going to execute . 1146   * @return Returns true if backend isReadEnabled() or isWriteEnabled() or 1147   * isReplaying() 1148   */ 1149  public boolean canAcceptTasks(long persistentConnectionId) 1150  { 1151    return canAcceptTasks(true, persistentConnectionId); 1152  } 1153 1154  private boolean canAcceptTasks(boolean isPersistentConnection, 1155      long persistentConnectionId) 1156  { 1157    // return isReadEnabled() || isWriteEnabled() || isReplaying(); 1158 boolean acceptTask = state == BackendState.READ_ENABLED_WRITE_DISABLED 1159        || state == BackendState.READ_ENABLED_WRITE_ENABLED 1160        || state == BackendState.READ_DISABLED_WRITE_ENABLED 1161        || state == BackendState.REPLAYING; 1162 1163    if (!acceptTask && isPersistentConnection) 1164    { 1165      // Check if the request is on one of our active persistent connections in 1166 // which case we have to execute it. 1167 synchronized (persistentConnections) 1168      { 1169        return persistentConnections.containsKey(new Long ( 1170            persistentConnectionId)); 1171      } 1172    } 1173    return acceptTask; 1174  } 1175 1176  /** 1177   * Cleans transactions and pending requests / tasks states 1178   */ 1179  private void cleanBackendStates() 1180  { 1181    activeTransactions.clear(); 1182    savepoints.clear(); 1183    pendingRequests.clear(); 1184    pendingTasks.clear(); 1185    setSchemaIsDirty(true, null); 1186    // make sure locks from old transactions are not carried over 1187 schema = null; 1188  } 1189 1190  /** 1191   * Sets the database backend state to disable. This state is just an 1192   * indication and it has no semantic effect. It is up to the request manager 1193   * (especially the load balancer) to ensure that no more requests are sent to 1194   * this backend. 1195   * 1196   * @return false if the backend was already in the disabled state, true 1197   * otherwise 1198   */ 1199  public synchronized boolean disable() 1200  { 1201    if (getStateValue() == BackendState.DISABLED) 1202    { 1203      return false; 1204    } 1205    setState(BackendState.DISABLED); 1206 1207    return true; 1208  } 1209 1210  /** 1211   * Disables the database backend for reads. This does not affect write ability 1212   */ 1213  public synchronized void disableRead() 1214  { 1215    if (isWriteEnabled()) 1216      setState(BackendState.READ_DISABLED_WRITE_ENABLED); 1217    else 1218      setState(BackendState.DISABLED); 1219  } 1220 1221  /** 1222   * Disables the database backend for writes. This does not affect read ability 1223   * although the backend will not be coherent anymore as soon as a write as 1224   * occured. This should be used in conjunction with a checkpoint to recover 1225   * missing writes. 1226   */ 1227  public synchronized void disableWrite() 1228  { 1229    if (isReadEnabled()) 1230      setState(BackendState.READ_ENABLED_WRITE_DISABLED); 1231    else 1232      setState(BackendState.DISABLED); 1233  } 1234 1235  /** 1236   * Enables the database backend for reads. This method should only be called 1237   * when the backend is synchronized with the others. 1238   */ 1239  public synchronized void enableRead() 1240  { 1241    if (isWriteEnabled()) 1242      setState(BackendState.READ_ENABLED_WRITE_ENABLED); 1243    else 1244      setState(BackendState.READ_ENABLED_WRITE_DISABLED); 1245  } 1246 1247  /** 1248   * Enables the database backend for writes. This method should only be called 1249   * when the backend is synchronized with the others. 1250   */ 1251  public synchronized void enableWrite() 1252  { 1253    // Remove last known checkpoint since backend will now be modified and no 1254 // more synchronized with the checkpoint. 1255 setLastKnownCheckpoint(null); 1256    if (isReadEnabled()) 1257      setState(BackendState.READ_ENABLED_WRITE_ENABLED); 1258    else 1259      setState(BackendState.READ_DISABLED_WRITE_ENABLED); 1260  } 1261 1262  /** 1263   * Returns the lastKnownCheckpoint value. 1264   * 1265   * @return Returns the lastKnownCheckpoint. 1266   */ 1267  public String getLastKnownCheckpoint() 1268  { 1269    return lastKnownCheckpoint; 1270  } 1271 1272  /** 1273   * Retrieve the state of the backend. 1274   * 1275   * @see SequoiaNotificationList#VIRTUALDATABASE_BACKEND_DISABLED 1276   * @see SequoiaNotificationList#VIRTUALDATABASE_BACKEND_RECOVERING 1277   * @see SequoiaNotificationList#VIRTUALDATABASE_BACKEND_BACKINGUP 1278   * @see SequoiaNotificationList#VIRTUALDATABASE_BACKEND_DISABLING 1279   * @see SequoiaNotificationList#VIRTUALDATABASE_BACKEND_ENABLED 1280   * @see SequoiaNotificationList#VIRTUALDATABASE_BACKEND_DISABLED 1281   * @return one of the above 1282   */ 1283  // FIXME should not return key for i18n but i18n translation instead 1284 public String getState() 1285  { 1286    switch (state) 1287    { 1288      case BackendState.READ_ENABLED_WRITE_DISABLED : 1289        return SequoiaNotificationList.VIRTUALDATABASE_BACKEND_ENABLED; 1290      case BackendState.READ_ENABLED_WRITE_ENABLED : 1291        return SequoiaNotificationList.VIRTUALDATABASE_BACKEND_ENABLED_WRITE; 1292      case BackendState.READ_DISABLED_WRITE_ENABLED : 1293        return SequoiaNotificationList.VIRTUALDATABASE_BACKEND_ENABLED_WRITE; 1294      case BackendState.DISABLING : 1295        return SequoiaNotificationList.VIRTUALDATABASE_BACKEND_DISABLING; 1296      case BackendState.BACKUPING : 1297        return SequoiaNotificationList.VIRTUALDATABASE_BACKEND_BACKINGUP; 1298      case BackendState.RESTORING : 1299        return SequoiaNotificationList.VIRTUALDATABASE_BACKEND_RECOVERING; 1300      case BackendState.REPLAYING : 1301        return SequoiaNotificationList.VIRTUALDATABASE_BACKEND_REPLAYING; 1302      case BackendState.DISABLED : 1303        return SequoiaNotificationList.VIRTUALDATABASE_BACKEND_DISABLED; 1304      case BackendState.UNKNOWN : 1305        return SequoiaNotificationList.VIRTUALDATABASE_BACKEND_UNKNOWN; 1306      default : 1307        throw new IllegalArgumentException ("Unknown backend state:" + state); 1308    } 1309  } 1310 1311  /** 1312   * Return the integer value corresponding to the state of the backend. The 1313   * values are defined in <code>BackendState</code> 1314   * 1315   * @return <tt>int</tt> value 1316   * @see BackendState 1317   */ 1318  public int getStateValue() 1319  { 1320    return state; 1321  } 1322 1323  /** 1324   * Returns the isBackuping value. 1325   * 1326   * @return Returns the isBackuping. 1327   */ 1328  public boolean isBackuping() 1329  { 1330    return state == BackendState.BACKUPING; 1331  } 1332 1333  /** 1334   * Is the backend completely disabled ? This usually means it has a known 1335   * state with a checkpoint associated to it. 1336   * 1337   * @return <code>true</code> if the backend is disabled 1338   */ 1339  public boolean isDisabled() 1340  { 1341    return state == BackendState.DISABLED; 1342  } 1343 1344  /** 1345   * Returns the isDisabling value. 1346   * 1347   * @return Returns the isDisabling. 1348   */ 1349  public boolean isDisabling() 1350  { 1351    return state == BackendState.DISABLING; 1352  } 1353 1354  /** 1355   * Tests if this backend is initialized 1356   * 1357   * @return <code>true</code> if this backend is initialized 1358   * @throws SQLException if an error occurs 1359   */ 1360  public synchronized boolean isInitialized() throws SQLException 1361  { 1362    if (connectionManagers.isEmpty()) 1363      throw new SQLException (Translate.get("backend.null.connection.manager", 1364          new String []{name, url})); 1365    Iterator iter = connectionManagers.values().iterator(); 1366    while (iter.hasNext()) 1367    { 1368      if (!((AbstractConnectionManager) iter.next()).isInitialized()) 1369        return false; 1370    } 1371    return true; 1372  } 1373 1374  /** 1375   * Is the backend accessible ? 1376   * 1377   * @return <tt>true</tt> if a jdbc connection is still possible from the 1378   * controller 1379   */ 1380  public synchronized boolean isJDBCConnected() 1381  { 1382    Connection con = null; 1383    try 1384    { 1385      if (connectionManagers.isEmpty()) 1386        throw new SQLException (Translate.get("backend.null.connection.manager", 1387            new String []{name, url})); 1388 1389      AbstractConnectionManager connectionManager; 1390      Iterator iter = connectionManagers.values().iterator(); 1391      connectionManager = (AbstractConnectionManager) iter.next(); 1392 1393      con = connectionManager.getConnectionFromDriver(); 1394      if (con == null) 1395      { 1396        return false; 1397      } 1398      con.createStatement().execute(this.connectionTestStatement); 1399      return true; 1400    } 1401    catch (Exception e) 1402    { 1403      String msg = Translate.get("loadbalancer.backend.unreacheable", name); 1404      logger.warn(msg, e); 1405      return false; 1406    } 1407    finally 1408    { 1409      if (con != null) 1410      { 1411        try 1412        { 1413          con.close(); 1414        } 1415        catch (SQLException e) 1416        { 1417          return false; 1418        } 1419      } 1420    } 1421  } 1422 1423  /** 1424   * Returns true if the backend cannot be used anymore 1425   * 1426   * @return Returns true if the backend was removed from activity by the load 1427   * balancer 1428   */ 1429  // TODO nobody uses this method. Should be removed 1430 public boolean isKilled() 1431  { 1432    return state == BackendState.UNKNOWN; 1433  } 1434 1435  /** 1436   * Tests if this backend is read enabled (active and synchronized). 1437   * 1438   * @return <code>true</code> if this backend is enabled. 1439   */ 1440  public synchronized boolean isReadEnabled() 1441  { 1442    return state == BackendState.READ_ENABLED_WRITE_DISABLED 1443        || state == BackendState.READ_ENABLED_WRITE_ENABLED; 1444  } 1445 1446  /** 1447   * Returns true if the backend is in the BackendState.RECOVERING state 1448   * 1449   * @return Returns true if the backend is restoring a dump 1450   */ 1451  // TODO nobody uses this method. Should be removed 1452 public boolean isRestoring() 1453  { 1454    return state == BackendState.RESTORING; 1455  } 1456 1457  /** 1458   * Returns true if the backend is in the BackendState.REPLAYING state 1459   * 1460   * @return Returns true if the backend is replaying the recovery 1461   */ 1462  public boolean isReplaying() 1463  { 1464    return state == BackendState.REPLAYING; 1465  } 1466 1467  /** 1468   * Tests if this backend is write enabled (active and synchronized). 1469   * 1470   * @return <code>true</code> if this backend is enabled. 1471   */ 1472  public synchronized boolean isWriteEnabled() 1473  { 1474    return state == BackendState.READ_ENABLED_WRITE_ENABLED 1475        || state == BackendState.READ_DISABLED_WRITE_ENABLED; 1476  } 1477 1478  /** 1479   * Returns the writeCanBeEnabled value. 1480   * 1481   * @return Returns the writeCanBeEnabled. 1482   */ 1483  public boolean isWriteCanBeEnabled() 1484  { 1485    return writeCanBeEnabled; 1486  } 1487 1488  /** 1489   * Sets the NotificationBroadcasterSupport that this backend can use to send 1490   * JMX notification. 1491   * 1492   * @param notificationBroadcaster a NotificationBroadcasterSupport 1493   */ 1494  public void setNotificationBroadcaster( 1495      NotificationBroadcasterSupport notificationBroadcaster) 1496  { 1497    this.notificationBroadcaster = notificationBroadcaster; 1498  } 1499 1500  /** 1501   * Notify JMX NotificationListener that the state of this DatabaseBackend has 1502   * changed. The emmitted Notification is a 1503   * <code>AttributeChangeNotification</code> with an attributeName set to 1504   * <code>"StateValue"</code> and an attributeType set to 1505   * <code>Integer</code> 1506   * 1507   * @param message a message about the backend state change 1508   * @param oldState the previous state of the backend 1509   * @param currentState the current state of the backend 1510   * @see BackendState 1511   * @see #getStateValue() 1512   */ 1513  private void notifyJMXStateChanged(String message, int oldState, 1514      int currentState) 1515  { 1516    try 1517    { 1518      Notification attrChangeNotification = new AttributeChangeNotification ( 1519          JmxConstants.getDatabaseBackendObjectName(vdb 1520              .getVirtualDatabaseName(), name), notificationSequence++, 1521          new Date ().getTime(), message, "StateValue", "Integer", //$NON-NLS-1$ //$NON-NLS-2$ 1522 new Integer (oldState), new Integer (currentState)); 1523      sendNotification(attrChangeNotification); 1524    } 1525    catch (MalformedObjectNameException e) 1526    { 1527      logger.warn("Unable to send JMX notification", e); 1528    } 1529  } 1530 1531  /** 1532   * Sends JMX notification 1533   * 1534   * @param type notification type 1535   * @see SequoiaNotificationList 1536   */ 1537  public void notifyJmx(String type) 1538  { 1539    notifyJmx(type, SequoiaNotificationList.NOTIFICATION_LEVEL_INFO, Translate 1540        .get(type, getName())); 1541  } 1542 1543  /** 1544   * Sends JMX error notification 1545   * 1546   * @param e <tt>Exception</tt> object. Only the message will be used 1547   * @param type notification type 1548   * @see SequoiaNotificationList 1549   */ 1550  public void notifyJmxError(String type, Exception e) 1551  { 1552    notifyJmx(type, SequoiaNotificationList.NOTIFICATION_LEVEL_ERROR, Translate 1553        .get(type, new String []{getName(), e.getMessage()})); 1554  } 1555 1556  /** 1557   * Sends a JMX notification. 1558   * 1559   * @param type Type of JMX notification 1560   * @param level unused parameter (will be deprecated) 1561   * @param message Message of the notification 1562   */ 1563  private void notifyJmx(String type, String level, String message) 1564  { 1565    try 1566    { 1567      Notification notification = (new Notification (type, JmxConstants 1568          .getDatabaseBackendObjectName(vdb.getVirtualDatabaseName(), name), 1569          notificationSequence++, message)); 1570 1571      sendNotification(notification); 1572    } 1573    catch (MalformedObjectNameException e) 1574    { 1575      logger.warn("Unable to send JMX notification", e); 1576    } 1577  } 1578 1579  /** 1580   * Sends a JMX notification. 1581   * 1582   * @param notification a JMX Notification to send 1583   */ 1584  private void sendNotification(Notification notification) 1585  { 1586    notificationBroadcaster.sendNotification(notification); 1587  } 1588 1589  /** 1590   * Notify the state of the backend has changed.<br /> 1591   * This notification triggers the update of the state of the backend stored in 1592   * the recovery log. 1593   */ 1594  private void notifyStateListener() 1595  { 1596    if (stateListener != null) 1597      stateListener.stateChanged(this); 1598  } 1599 1600  /** 1601   * Sets the stateListener value. 1602   * 1603   * @param stateListener The stateListener to set. 1604   */ 1605  public void setStateListener(BackendStateListener stateListener) 1606  { 1607    this.stateListener = stateListener; 1608  } 1609 1610  /** 1611   * This is used when the backend must be disabled but currently open 1612   * transactions must terminate. This is a transitional state. When disabling 1613   * is complete the caller must set the backend state to disabled. 1614   * <p> 1615   * Reads are no more allowed on the backend and the state is updated so that 1616   * isReadEnabled() returns false. 1617   * 1618   * @see #disable() 1619   * @see #isReadEnabled() 1620   * @deprecated not used anymore. Please use the setState method instead 1621   */ 1622  // TODO nobody uses this method. Should be removed 1623 public void setDisabling() 1624  { 1625    setState(BackendState.DISABLING); 1626  } 1627 1628  /** 1629   * setLastKnownCheckpoint for this backend 1630   * 1631   * @param checkpoint the checkpoint 1632   */ 1633  public void setLastKnownCheckpoint(String checkpoint) 1634  { 1635    this.lastKnownCheckpoint = checkpoint; 1636    notifyStateListener(); // triggers recovery log update 1637 } 1638 1639  /** 1640   * Set the state of a backend 1641   * 1642   * @param state see BackendState for a possible list of the different state 1643   * @see org.continuent.sequoia.common.jmx.management.BackendState 1644   */ 1645  // FIXME should use a type-safe enum to represent backend state instead of 1646 // ints 1647 public synchronized void setState(int state) 1648  { 1649    switch (state) 1650    { 1651      case BackendState.UNKNOWN : 1652        lastKnownCheckpoint = null; 1653        break; 1654      case BackendState.RESTORING : 1655      case BackendState.REPLAYING : 1656        cleanBackendStates(); 1657        break; 1658      case BackendState.READ_ENABLED_WRITE_DISABLED : 1659      case BackendState.READ_ENABLED_WRITE_ENABLED : 1660      case BackendState.READ_DISABLED_WRITE_ENABLED : 1661      case BackendState.DISABLING : 1662      case BackendState.BACKUPING : 1663      case BackendState.DISABLED : 1664        break; 1665      default : 1666        throw new IllegalArgumentException ("Unknown backend state:" + state); 1667    } 1668    int oldState = this.state; 1669    this.state = state; 1670    int currentState = this.state; 1671    if (logger.isDebugEnabled()) 1672      logger.debug(Translate.get("backend.state.changed", new String []{name, 1673          getState()})); 1674    endUserLogger.info(Translate.get("backend.state.changed", new String []{ 1675        name, BackendState.description(currentState)})); 1676 1677    notifyStateListener(); 1678    notifyJMXStateChanged(Translate.get(getState(), name), oldState, 1679        currentState); 1680  } 1681 1682  // 1683 // Schema manipulation 1684 // 1685 1686  /** 1687   * Checks that the current database schema is compatible with all schema 1688   * gathered from each connection manager. 1689   * <p> 1690   * If no schema has been defined, the first gathered schema is used as the 1691   * current database schema. 1692   * <p> 1693   * For each schema that is not compatible with the current schema, a warning 1694   * is issued on the logger for that backend 1695   * 1696   * @param c optional connection from which the schema should be fetched (null 1697   * if not applicable) 1698   * @return true if compatible, false otherwise 1699   */ 1700  public synchronized boolean checkDatabaseSchema(Connection c) 1701  { 1702    if (logger.isDebugEnabled()) 1703      logger.debug(Translate.get("backend.dynamic.schema", 1704          DatabaseBackendSchemaConstants 1705              .fetchStoredProcedures(dynamicPrecision))); 1706 1707    boolean checked = true; 1708    if (c == null) 1709    { 1710      AbstractConnectionManager connectionMananger; 1711      Iterator iter = connectionManagers.values().iterator(); 1712      while (iter.hasNext()) 1713      { 1714        connectionMananger = (AbstractConnectionManager) iter.next(); 1715 1716        // Gather the database schema from this connection manager 1717 DatabaseBackendMetaData meta = new DatabaseBackendMetaData( 1718            connectionMananger, logger, vdb.getDynamicDatabaseSchema(), vdb 1719                .getVirtualDatabaseName()); 1720 1721        DatabaseSchema metaSchema; 1722        try 1723        { 1724          if (logger.isDebugEnabled()) 1725            logger.debug(Translate.get("backend.gathering.database.schema")); 1726          metaSchema = meta.getDatabaseSchema(); 1727        } 1728        catch (SQLException e) 1729        { 1730          if (logger.isWarnEnabled()) 1731            logger.warn(Translate.get("backend.gather.schema.failed", e)); 1732          return false; 1733        } 1734        if (schema == null) 1735        { 1736          if (logger.isDebugEnabled()) 1737            logger.debug(Translate.get("backend.use.gathered.schema.as.new")); 1738          schema = metaSchema; 1739        } 1740        else 1741        { 1742          if (logger.isInfoEnabled()) 1743            logger.info(Translate.get("backend.check.schema.compatibility")); 1744          if (schema.isCompatibleSubset(metaSchema)) 1745            logger.info(Translate.get("backend.schema.compatible.for.login", 1746                connectionMananger.getLogin())); 1747          else 1748          { 1749            checked = false; 1750            logger.warn(Translate.get( 1751                "backend.schema.not.compatible.for.login", connectionMananger 1752                    .getLogin())); 1753          } 1754        } 1755      } 1756    } 1757    else 1758    { // Fetch schema from given connection 1759 try 1760      { 1761        schema = new DatabaseSQLMetaData(logger, c, vdb 1762            .getDynamicDatabaseSchema()).createDatabaseSchema(vdb 1763            .getVirtualDatabaseName()); 1764      } 1765      catch (SQLException e) 1766      { 1767        if (logger.isInfoEnabled()) 1768          logger.info("Failed to fetch schema from given connection " + c, e); 1769        return checkDatabaseSchema(null); 1770      } 1771    } 1772 1773    setSchemaIsDirty(false, null); 1774    return checked; 1775  } 1776 1777  /** 1778   * Returns the schema of this database. 1779   * 1780   * @return the schema of this database. Returns <code>null</code> if the 1781   * schema has not been set. 1782   * @see #setDatabaseSchema(DatabaseSchema) 1783   */ 1784  public synchronized DatabaseSchema getDatabaseSchema() 1785  { 1786    if (schemaIsDirty) 1787      refreshSchema(null); 1788    return schema; 1789  } 1790 1791  /** 1792   * Get the Database static metadata from this backend using a connection from 1793   * the first available connection manager. 1794   * 1795   * @return Static metadata information 1796   */ 1797  public MetadataContainer getDatabaseStaticMetadata() 1798  { 1799    AbstractConnectionManager connectionMananger; 1800    Iterator iter = connectionManagers.values().iterator(); 1801    if (iter.hasNext()) 1802    { 1803      connectionMananger = (AbstractConnectionManager) iter.next(); 1804      // Gather the static metadata from the first connection manager 1805 DatabaseBackendMetaData meta = new DatabaseBackendMetaData( 1806          connectionMananger, logger, vdb.getDynamicDatabaseSchema(), vdb 1807              .getVirtualDatabaseName()); 1808      try 1809      { 1810        return meta.retrieveDatabaseMetadata(); 1811      } 1812      catch (SQLException e) 1813      { 1814        return null; 1815      } 1816    } 1817    else 1818      return null; 1819  } 1820 1821  private static final Object DYNAMIC_PRECISION_SYNC = new Object (); 1822 1823  /** 1824   * Get the dynamic schema fetching precision (for backward compatibility only, 1825   * now found in DynamicDatabaseSchema) 1826   * 1827   * @return Returns the dynamicPrecision. 1828   * @see DynamicDatabaseSchema 1829   */ 1830  public int getDynamicPrecision() 1831  { 1832    synchronized (DYNAMIC_PRECISION_SYNC) 1833    { 1834      if (dynamicPrecision != -1) // Already set 1835 return dynamicPrecision; 1836 1837      // Compute dynamic precision 1838 dynamicPrecision = DatabaseBackendSchemaConstants.FetchTables; 1839      DynamicDatabaseSchema dynamicSchema = vdb.getDynamicDatabaseSchema(); 1840      if (dynamicSchema.gatherSystemTables()) 1841        dynamicPrecision += DatabaseBackendSchemaConstants.FetchSystemTables; 1842      if (dynamicSchema.useStoredProcedures()) 1843        dynamicPrecision += DatabaseBackendSchemaConstants.FetchStoredProcedures; 1844      if (dynamicSchema.useViews()) 1845        dynamicPrecision += DatabaseBackendSchemaConstants.FetchViews; 1846      return dynamicPrecision; 1847    } 1848  } 1849 1850  /** 1851   * Returns the schemaName value. 1852   * 1853   * @return Returns the schemaName. 1854   */ 1855  public String getSchemaName() 1856  { 1857    return vdb.getDynamicDatabaseSchema().getSchemaName(); 1858  } 1859 1860  /** 1861   * Get all the names of tables of this database <b>NOTE</b>: The returned 1862   * collection will contain two entries per actual table: one with the table 1863   * name alone, the other prefixed by the schema name + ".". 1864   * 1865   * @see org.continuent.sequoia.controller.sql.schema.DatabaseSchema#addTable(DatabaseTable) 1866   * @return <code>Collection</code> of <code>DatabaseTable</code> 1867   */ 1868  public Collection getTables() 1869  { 1870    DatabaseSchema schemaPtr = getDatabaseSchema(); 1871    if (schemaPtr == null) 1872      throw new NullPointerException (Translate.get("backend.schema.not.set")); 1873    return schemaPtr.getTables().values(); 1874  } 1875 1876  /** 1877   * Returns <code>true</code> if this backend has the given table in its 1878   * schema. The caller must ensure that the database schema has been defined, 1879   * using the {@link #setDatabaseSchema(DatabaseSchema)}or 1880   * {@link #checkDatabaseSchema(Connection)} 1881   * 1882   * @param table The table name to look for 1883   * @return <code>true</code> if tables is found in the schema 1884   */ 1885  public boolean hasTable(String table) 1886  { 1887    DatabaseSchema schemaPtr = getDatabaseSchema(); 1888    if (schemaPtr == null) 1889      throw new NullPointerException (Translate.get("backend.schema.not.set")); 1890 1891    return schemaPtr.hasTable(table); 1892  } 1893 1894  /** 1895   * Returns <code>true</code> if this backend has the given list of tables in 1896   * its schema. The caller must ensure that the database schema has been 1897   * defined, using the {@link #setDatabaseSchema(DatabaseSchema)}or 1898   * {@link #checkDatabaseSchema(Connection)}methods. 1899   * 1900   * @param tables the list of table names (<code>Collection</code> of 1901   * <code>String</code>) to look for 1902   * @return <code>true</code> if all the tables are found 1903   */ 1904  public boolean hasTables(Collection tables) 1905  { 1906    DatabaseSchema schemaPtr = getDatabaseSchema(); 1907    if (schemaPtr == null) 1908      throw new NullPointerException (Translate.get("backend.schema.not.set")); 1909 1910    if (tables == null) 1911      throw new IllegalArgumentException (Translate.get("backend.null.tables")); 1912 1913    for (Iterator iter = tables.iterator(); iter.hasNext();) 1914    { 1915      if (!schemaPtr.hasTable((String ) iter.next())) 1916        return false; 1917    } 1918    return true; 1919  } 1920 1921  /** 1922   * Returns <code>true</code> if this backend has the given stored procedure 1923   * in its schema. The caller must ensure that the database schema has been 1924   * defined, using the {@link #setDatabaseSchema(DatabaseSchema)}or 1925   * {@link #checkDatabaseSchema(Connection)} 1926   * 1927   * @param procedureName The stored procedure name to look for 1928   * @param nbOfParameters number of parameters of the stored procecdure 1929   * @return <code>true</code> if the procedure has been found 1930   */ 1931  public boolean hasStoredProcedure(String procedureName, int nbOfParameters) 1932  { 1933    DatabaseSchema schemaPtr = getDatabaseSchema(); 1934    if (schemaPtr == null) 1935      throw new NullPointerException (Translate.get("backend.schema.not.set")); 1936 1937    return schemaPtr.hasProcedure(procedureName, nbOfParameters); 1938  } 1939 1940  /** 1941   * Returns the schemaIsDirty value. 1942   * 1943   * @return Returns true if the backend database schema is dirty and needs a 1944   * refresh. 1945   */ 1946  // TODO nobody uses this method. Should be removed 1947 public boolean isSchemaDirty() 1948  { 1949    return schemaIsDirty; 1950  } 1951 1952  /** 1953   * Erase the current schema and force a re-fetch of all the meta data 1954   * 1955   * @param c optional connection from which the schema should be fetched (null 1956   * if not applicable) 1957   */ 1958  private synchronized void refreshSchema(Connection c) 1959  { 1960    DatabaseSchema oldSchema = schema; 1961    setDatabaseSchema(null); 1962    checkDatabaseSchema(c); // set dirty to false as well 1963 // Schema is null if we failed to refresh it 1964 if ((schema != null) && (oldSchema != null)) 1965    { 1966      schema.setLocks(oldSchema); 1967    } 1968  } 1969 1970  /** 1971   * Sets the database schema. 1972   * 1973   * @param databaseSchema the schema to set 1974   * @see #getDatabaseSchema() 1975   */ 1976  public synchronized void setDatabaseSchema(DatabaseSchema databaseSchema) 1977  { 1978    schema = databaseSchema; 1979  } 1980 1981  /** 1982   * Sets the schemaIsDirty value if the backend schema needs to be refreshed. 1983   * 1984   * @param schemaIsDirty The schemaIsDirty to set. 1985   * @param request optional request information used to retrieve the connection 1986   * from which the schema should be fetched (null if not applicable) 1987   */ 1988  public void setSchemaIsDirty(boolean schemaIsDirty, AbstractRequest request) 1989  { 1990    if (request == null) 1991    { 1992      this.schemaIsDirty = schemaIsDirty; 1993      return; 1994    } 1995 1996    // Try to retrieve the connection corresponding to the persistent connection 1997 // or transaction if applicable 1998 Connection c = null; 1999    PooledConnection pc = null; 2000    if (request.isPersistentConnection()) 2001    { 2002      AbstractConnectionManager cm = getConnectionManager(request.getLogin()); 2003      if (cm != null) 2004        try 2005        { 2006          pc = cm.retrieveConnectionInAutoCommit(request); 2007          if (pc != null) 2008            c = pc.getConnection(); 2009        } 2010        catch (UnreachableBackendException ignore) 2011        { 2012        } 2013    } 2014    else if (!request.isAutoCommit()) 2015    { 2016      AbstractConnectionManager cm = getConnectionManager(request.getLogin()); 2017      if (cm != null) 2018        pc = cm.retrieveConnectionForTransaction(request.getTransactionId()); 2019      if (pc != null) 2020        c = pc.getConnection(); 2021    } 2022 2023    if (c == null) 2024      this.schemaIsDirty = schemaIsDirty; 2025    else if (schemaIsDirty) 2026      // refresh schema right away 2027 refreshSchema(c); 2028  } 2029 2030  /** 2031   * Sets the schemaIsDirty value to true if the backend schema needs to be 2032   * refreshed (no semantic info or DDL executed by stored procedure). 2033   * 2034   * @param proc stored procedure executing (preferably with semantic 2035   * information) 2036   */ 2037  public void setSchemaIsDirtyIfNeeded(StoredProcedure proc) 2038  { 2039    SemanticBehavior semantic = proc.getSemantic(); 2040    if ((semantic == null) || semantic.altersDatabaseSchema()) 2041      setSchemaIsDirty(true, proc); 2042  } 2043 2044  /** 2045   * Update the DatabaseBackend schema definition according to the successful 2046   * execution of the provided request. Note that the schema is only updated it 2047   * the provided request is a DDL statement. 2048   * <p> 2049   * TODO: a full refresh is forced on CREATE request to be sure to properly 2050   * handle foreign keys (see SEQUOIA-581). An improvement would be not to 2051   * refresh the whole schema on each create request - just add the new table 2052   * and fetch its dependencies (foreign keys) by reloading exported keys of 2053   * existing tables (see SEQUOIA-xxx). Similar optimization may be done on 2054   * ALTER requests but it would require to be able to parse the request - 2055   * currently we just force a full schema refresh. Note that concerning DROP 2056   * request, we remove the table from dependending tables of other tables in 2057   * addition to the removal of the table from the schema (given t1 and t2 2058   * tables, if t2 has a fk referencing t1, the table t1 will have the table t2 2059   * in its dependending tables). In case of rollback, these modification will 2060   * be cancelled because the schema will be refreshed. 2061   * 2062   * @param request the request that possibly updates the schema 2063   */ 2064  public void updateDatabaseBackendSchema(AbstractWriteRequest request) 2065  { 2066    getDatabaseSchema().updateDatabaseSchema(request, logger, 2067        vdb.getRequestManager(), null, this); 2068  } 2069 2070  // 2071 // Rewriting rules 2072 // 2073 2074  /** 2075   * Add a <code>AbstractRewritingRule</code> at the end of the rule list. 2076   * 2077   * @param rule a AbstractRewritingRule 2078   */ 2079  public void addRewritingRule(AbstractRewritingRule rule) 2080  { 2081    if (rewritingRules == null) 2082      rewritingRules = new ArrayList (); 2083    if (logger.isDebugEnabled()) 2084      logger.debug(Translate.get("backend.rewriting.rule.add", new String []{ 2085          rule.getQueryPattern(), rule.getRewrite()})); 2086    rewritingRules.add(rule); 2087  } 2088 2089  /** 2090   * Rewrite the current query according to the rewriting rules. 2091   * 2092   * @param sqlQuery request to rewrite 2093   * @return the rewritten SQL query according to rewriting rules. 2094   */ 2095  public String rewriteQuery(String sqlQuery) 2096  { 2097    if (rewritingRules == null) 2098      return sqlQuery; 2099    int size = rewritingRules.size(); 2100    for (int i = 0; i < size; i++) 2101    { 2102      AbstractRewritingRule rule = (AbstractRewritingRule) rewritingRules 2103          .get(i); 2104      sqlQuery = rule.rewrite(sqlQuery); 2105      if (rule.hasMatched()) 2106      { // Rule matched, query rewriten 2107 if (logger.isDebugEnabled()) 2108          logger.debug(Translate.get("backend.rewriting.query", sqlQuery)); 2109        if (rule.isStopOnMatch()) 2110          break; // Ok, stop here. 2111 } 2112    } 2113    return sqlQuery; 2114  } 2115 2116  /* 2117   * Debug/Monitoring 2118   */ 2119 2120  /** 2121   * Adds a pending request to this backend.<br /> 2122   * Method is synchronized on <code>pendingRequests</code> field. 2123   * 2124   * @param request the request to add 2125   * @see #pendingRequests 2126   */ 2127  public void addPendingReadRequest(AbstractRequest request) 2128  { 2129    synchronized (pendingRequests) 2130    { 2131      totalRequest++; 2132      totalReadRequest++; 2133      pendingRequests.add(request); 2134    } 2135  } 2136 2137  /** 2138   * Adds a pending request to this backend.<br /> 2139   * Method is synchronized on <code>pendingTasks</code> field. 2140   * 2141   * @param task the task to add 2142   * @see #pendingTasks 2143   */ 2144  public void addPendingTask(AbstractTask task) 2145  { 2146    synchronized (pendingTasks) 2147    { 2148      totalTasks++; 2149      pendingTasks.add(task); 2150    } 2151  } 2152 2153  /** 2154   * Adds a pending write request to this backend.<br /> 2155   * Method is synchronized on <code>pendingRequests</code> field. 2156   * 2157   * @param request the request to add 2158   * @see #pendingRequests 2159   */ 2160  public void addPendingWriteRequest(AbstractRequest request) 2161  { 2162    synchronized (pendingRequests) 2163    { 2164      totalRequest++; 2165      totalWriteRequest++; 2166      pendingRequests.add(request); 2167    } 2168  } 2169 2170  /** 2171   * @return Returns the activeTransactions. 2172   */ 2173  public ArrayList getActiveTransactions() 2174  { 2175    return activeTransactions; 2176  } 2177 2178  /** 2179   * Get data about this backend. Format is: 2180   * 2181   * <pre> 2182   * data[0] = this.name; 2183   * data[1] = this.driverClassName; 2184   * data[2] = this.url; 2185   * data[3] = String.valueOf(this.activeTransactions.size()); 2186   * data[4] = String.valueOf(this.pendingRequests.size()); 2187   * data[5] = String.valueOf(this.isReadEnabled()); 2188   * data[6] = String.valueOf(this.isWriteEnabled()); 2189   * data[7] = String.valueOf(this.isInitialized()); 2190   * data[8] = String.valueOf(this.schemaIsStatic); 2191   * data[9] = String.valueOf(this.connectionManagers.size()); 2192   * data[10] = String.valueOf(getTotalActiveConnections()); 2193   * data[11] = String.valueOf(totalRequest); 2194   * data[12] = String.valueOf(totalTransactions); 2195   * data[13] = lastKnownCheckpoint; 2196   *</pre> 2197   * 2198   * @return an array of strings 2199   */ 2200  public String [] getBackendData() 2201  { 2202    String [] data = new String [14]; 2203    data[0] = this.name; 2204    data[1] = this.driverClassName; 2205    data[2] = this.url; 2206    data[3] = String.valueOf(this.activeTransactions.size()); 2207    data[4] = String.valueOf(this.pendingRequests.size()); 2208    data[5] = String.valueOf(this.isReadEnabled()); 2209    data[6] = String.valueOf(this.isWriteEnabled()); 2210    try 2211    { 2212      data[7] = String.valueOf(this.isInitialized()); 2213    } 2214    catch (Exception e) 2215    { 2216      data[7] = "unknown"; 2217    } 2218 2219    // static schema is now deprecated (kept for backward compatibility) 2220 data[8] = String.valueOf(false); 2221 2222    data[9] = String.valueOf(this.connectionManagers.size()); 2223    data[10] = String.valueOf(getTotalActiveConnections()); 2224    data[11] = String.valueOf(totalRequest); 2225    data[12] = String.valueOf(totalTransactions); 2226    if (lastKnownCheckpoint == null || lastKnownCheckpoint.equalsIgnoreCase("")) 2227      data[13] = "<unknown>"; 2228    else 2229      data[13] = lastKnownCheckpoint; 2230    return data; 2231  } 2232 2233  /** 2234   * Get the statistics of the backend. 2235   * 2236   * @return a BackendStatistics 2237   */ 2238  public BackendStatistics getBackendStats() 2239  { 2240    BackendStatistics stats = new BackendStatistics(); 2241    stats.setBackendName(name); 2242    stats.setDriverClassName(driverClassName); 2243    stats.setUrl(url); 2244    stats.setNumberOfActiveTransactions(activeTransactions.size()); 2245    stats.setNumberOfPendingRequests(pendingRequests.size()); 2246    stats.setNumberOfPersistentConnections(persistentConnections.size()); 2247    stats.setReadEnabled(isReadEnabled()); 2248    stats.setWriteEnabled(isWriteEnabled()); 2249    String initializationStatus = "<unknown>"; 2250    try 2251    { 2252      initializationStatus = String.valueOf(this.isInitialized()); 2253    } 2254    catch (Exception e) 2255    { 2256    } 2257    stats.setInitializationStatus(initializationStatus); 2258    stats.setNumberOfConnectionManagers(connectionManagers.size()); 2259    stats.setNumberOfTotalActiveConnections(getTotalActiveConnections()); 2260    stats.setNumberOfTotalRequests(totalRequest); 2261    stats.setNumberOfTotalTransactions(totalTransactions); 2262    if (lastKnownCheckpoint == null || lastKnownCheckpoint.equalsIgnoreCase("")) 2263      stats.setLastKnownCheckpoint("<unknown>"); 2264    else 2265      stats.setLastKnownCheckpoint(lastKnownCheckpoint); 2266    return stats; 2267  } 2268 2269  /** 2270   * Returns the list of pending requests for this backend. 2271   * 2272   * @return <code>Vector</code> of <code>AbstractRequests</code> or 2273   * <code>AbstractTask</code> objects 2274   */ 2275  public Vector getPendingRequests() 2276  { 2277    return pendingRequests; 2278  } 2279 2280  /** 2281   * Returns the list of pending requests for this backend. 2282   * 2283   * @param count number of requests to retrieve, if 0, return all. 2284   * @param fromFirst count the request from first if true, or from last if 2285   * false 2286   * @param clone should clone the pending request if true, block it if false 2287   * @return <code>ArrayList</code> of <code>String</code> description of 2288   * each request. 2289   */ 2290  public ArrayList getPendingRequestsDescription(int count, boolean fromFirst, 2291      boolean clone) 2292  { 2293    int size = pendingRequests.size(); 2294    int limit = (count == 0 || count > size) ? size : Math.min(size, count); 2295    ArrayList list = new ArrayList (limit); 2296    int start = (fromFirst) ? 0 : Math.min(limit - count, 0); 2297    if (!clone) 2298    { 2299      synchronized (pendingRequests) 2300      { 2301        for (int i = start; i < limit; i++) 2302          list.add(pendingRequests.get(i).toString()); 2303      } 2304      return list; 2305    } 2306    else 2307    { 2308      Vector cloneVector = (Vector ) pendingRequests.clone(); 2309      for (int i = start; i < limit; i++) 2310        list.add(cloneVector.get(i).toString()); 2311      return list; 2312    } 2313  } 2314 2315  /** 2316   * Get the total number of active connections for this backend 2317   * 2318   * @return number of active connections for all 2319   * <code>AbstractConnectionManager</code> connected to this backend 2320   */ 2321  public long getTotalActiveConnections() 2322  { 2323    int activeConnections = 0; 2324    Iterator iter = connectionManagers.keySet().iterator(); 2325    while (iter.hasNext()) 2326      activeConnections += ((AbstractConnectionManager) connectionManagers 2327          .get(iter.next())).getCurrentNumberOfConnections(); 2328    return activeConnections; 2329  } 2330 2331  /** 2332   * Returns the total number of transactions executed by this backend. 2333   * 2334   * @return Total number of transactions. 2335   */ 2336  public int getTotalTransactions() 2337  { 2338    return totalTransactions; 2339  } 2340 2341  /** 2342   * Returns the total number of read requests executed by this backend. 2343   * 2344   * @return Returns the totalReadRequest. 2345   */ 2346  public int getTotalReadRequest() 2347  { 2348    return totalReadRequest; 2349  } 2350 2351  /** 2352   * Returns the total number of write requests executed by this backend. 2353   * 2354   * @return Returns the totalWriteRequest. 2355   */ 2356  public int getTotalWriteRequest() 2357  { 2358    return totalWriteRequest; 2359  } 2360 2361  /** 2362   * Returns the total number of requests executed by this backend. 2363   * 2364   * @return Returns the totalRequest. 2365   */ 2366  public int getTotalRequest() 2367  { 2368    return totalRequest; 2369  } 2370 2371  /** 2372   * Removes a pending request from this backend. Note that the underlying 2373   * vector is synchronized. 2374   * 2375   * @param request the request to remove 2376   * @return <code>true</code> if the request has been found and removed 2377   */ 2378  public boolean removePendingRequest(AbstractRequest request) 2379  { 2380    return pendingRequests.remove(request); 2381  } 2382 2383  /** 2384   * Removes a pending task from this backend. Note that the underlying vector 2385   * is synchronized. 2386   * 2387   * @param task the task to remove 2388   * @return <code>true</code> if the task has been found and removed 2389   */ 2390  public boolean removePendingTask(AbstractTask task) 2391  { 2392    return pendingTasks.remove(task); 2393  } 2394 2395  // 2396 // Getters/Setters 2397 // 2398 2399  /** 2400   * Returns the databaseProductName value. 2401   * 2402   * @return Returns the databaseProductName. 2403   */ 2404  public String getDatabaseProductName() 2405  { 2406    return driverCompliance.getDatabaseProductName(); 2407  } 2408 2409  /** 2410   * @return the driver compliance to Sequoia requirements. 2411   */ 2412  public DriverCompliance getDriverCompliance() 2413  { 2414    return driverCompliance; 2415  } 2416 2417  /** 2418   * Returns the driver path. 2419   * 2420   * @return the driver path 2421   */ 2422  public String getDriverPath() 2423  { 2424    return driverPath; 2425  } 2426 2427  /** 2428   * Returns the database native JDBC driver class name. 2429   * 2430   * @return the driver class name 2431   */ 2432  public String getDriverClassName() 2433  { 2434    return driverClassName; 2435  } 2436 2437  /** 2438   * Returns the backend logical name. 2439   * 2440   * @return the backend logical name 2441   */ 2442  public String getName() 2443  { 2444    return name; 2445  } 2446 2447  /** 2448   * Return the sql short form length to use when reporting an error. 2449   * 2450   * @return sql short form length 2451   * @see org.continuent.sequoia.controller.requests.AbstractRequest#getSqlShortForm(int) 2452   */ 2453  public int getSqlShortFormLength() 2454  { 2455    return sqlShortFormLength; 2456  } 2457 2458  /** 2459   * Returns the JDBC URL used to access the database. 2460   * 2461   * @return a JDBC URL 2462   */ 2463  public String getURL() 2464  { 2465    return url; 2466  } 2467 2468  /** 2469   * Returns the taskQueues value. 2470   * 2471   * @return Returns the taskQueues. 2472   */ 2473  public BackendTaskQueues getTaskQueues() 2474  { 2475    return taskQueues; 2476  } 2477 2478  /** 2479   * Sets the taskQueues value. 2480   * 2481   * @param taskQueues The taskQueues to set. 2482   */ 2483  public void setTaskQueues(BackendTaskQueues taskQueues) 2484  { 2485    this.taskQueues = taskQueues; 2486  } 2487 2488  /** 2489   * Returns the virtual database name this backend belongs to. 2490   * 2491   * @return Returns the virtual database name. 2492   */ 2493  public String getVirtualDatabaseName() 2494  { 2495    return vdb.getVirtualDatabaseName(); 2496  } 2497 2498  // 2499 // XML mapping 2500 // 2501 2502  /** 2503   * Get xml information about this backend. 2504   * 2505   * @return xml formatted information on this database backend. 2506   */ 2507  public synchronized String getXml() 2508  { 2509    // escape & XML entity and replace it by &amp; so that the 2510 // url attribute is XML compliant 2511 String escapedUrl = url.replaceAll("&", "&amp;"); 2512 2513    StringBuffer info = new StringBuffer (); 2514    info.append("<" + DatabasesXmlTags.ELT_DatabaseBackend + " " 2515        + DatabasesXmlTags.ATT_name + "=\"" + name + "\" " 2516        + DatabasesXmlTags.ATT_driver + "=\"" + driverClassName + "\" " 2517        + DatabasesXmlTags.ATT_url + "=\"" + escapedUrl + "\" " 2518        + DatabasesXmlTags.ATT_connectionTestStatement + "=\"" 2519        + connectionTestStatement + "\">"); 2520 2521    if (rewritingRules != null) 2522    { 2523      int size = rewritingRules.size(); 2524      for (int i = 0; i < size; i++) 2525        info.append(((AbstractRewritingRule) rewritingRules.get(i)).getXml()); 2526    } 2527    if (connectionManagers != null) 2528    { 2529      if (connectionManagers.isEmpty() == false) 2530      { 2531        AbstractConnectionManager connectionManager; 2532        Iterator iter = connectionManagers.values().iterator(); 2533        while (iter.hasNext()) 2534        { 2535          connectionManager = (AbstractConnectionManager) iter.next(); 2536          info.append(connectionManager.getXml()); 2537        } 2538      } 2539    } 2540    info.append("</" + DatabasesXmlTags.ELT_DatabaseBackend + ">"); 2541    return info.toString(); 2542  } 2543 2544  /** 2545   * Sets the sqlShortFormLength value. 2546   * 2547   * @param sqlShortFormLength The sqlShortFormLength to set. 2548   */ 2549  public void setSqlShortFormLength(int sqlShortFormLength) 2550  { 2551    this.sqlShortFormLength = sqlShortFormLength; 2552  } 2553 2554  /** 2555   * String description 2556   * 2557   * @return a string description of the backend. 2558   */ 2559  public String toString() 2560  { 2561    return "Backend: Name[" + this.name + "] State[" + this.state 2562        + "] JDBCConnected[" + isJDBCConnected() + "] ActiveTransactions[" 2563        + activeTransactions.size() + "] PersistentConnections[" 2564        + persistentConnections + "] PendingRequests[" + pendingRequests.size() 2565        + "]"; 2566  } 2567 2568  // 2569 // Worker threads management 2570 // 2571 2572  /** 2573   * Return the first available BackendWorkerThread. This should only be used 2574   * for notification of request abort. 2575   * 2576   * @return a BackendWorkerThread or null if none exists 2577   */ 2578  public BackendWorkerThread getBackendWorkerThreadForNotification() 2579  { 2580    if ((workerThreads == null) || workerThreads.isEmpty()) 2581      return null; 2582    return (BackendWorkerThread) workerThreads.get(0); 2583  } 2584 2585  /** 2586   * Returns the nbOfWorkerThreads value. 2587   * 2588   * @return Returns the nbOfWorkerThreads. 2589   */ 2590  public int getNbOfWorkerThreads() 2591  { 2592    return nbOfWorkerThreads; 2593  } 2594 2595  /** 2596   * Sets the nbOfWorkerThreads value. 2597   * 2598   * @param nbOfWorkerThreads The nbOfWorkerThreads to set. 2599   */ 2600  // TODO nobody uses this method. Should be removed 2601 public void setNbOfWorkerThreads(int nbOfWorkerThreads) 2602  { 2603    this.nbOfWorkerThreads = nbOfWorkerThreads; 2604  } 2605 2606  /** 2607   * Start a new Deadlock Detection Thread (throws a RuntimeException if called 2608   * twice without stopping the thread before the second call). 2609   * 2610   * @param vdb the virtual database the backend is attached to 2611   */ 2612  public void startDeadlockDetectionThread(VirtualDatabase vdb) 2613  { 2614    taskQueues.startDeadlockDetectionThread(vdb); 2615 2616  } 2617 2618  /** 2619   * Start the BackendWorkerThreads for this backend. 2620   * 2621   * @param loadBalancer load balancer requesting the activation 2622   */ 2623  public void startWorkerThreads(AbstractLoadBalancer loadBalancer) 2624  { 2625    taskQueues.setAllowTasksToBePosted(true); 2626    synchronized (workerThreadSync) 2627    { 2628      if (logger.isDebugEnabled()) 2629        logger.debug(Translate.get( 2630            "loadbalancer.backend.workerthread.starting", new String []{ 2631                String.valueOf(nbOfWorkerThreads), name})); 2632 2633      if (workerThreads == null) 2634        workerThreads = new ArrayList (); 2635      // Create worker threads 2636 for (int i = 0; i < nbOfWorkerThreads; i++) 2637      { 2638        BackendWorkerThread thread = new BackendWorkerThread(this, loadBalancer); 2639        workerThreads.add(thread); 2640        // Dedicate the first thread to commit/rollback operations 2641 thread.setPlayCommitRollbackOnly(i == 0); 2642        thread.start(); 2643      } 2644    } 2645  } 2646 2647  /** 2648   * Terminate all worker threads. 2649   */ 2650  public void terminateWorkerThreads() 2651  { 2652    synchronized (workerThreadSync) 2653    { 2654      if (workerThreads == null) 2655        return; 2656 2657      // Terminate worker threads by posting a kill task for each of them 2658 int size = workerThreads.size(); 2659 2660      if (logger.isDebugEnabled()) 2661        logger.debug(Translate.get( 2662            "loadbalancer.backend.workerthread.stopping", new String []{ 2663                String.valueOf(size), name})); 2664 2665      for (int i = 0; i < size; i++) 2666      { 2667        KillThreadTask killBlockingThreadTask = new KillThreadTask(1, 1); 2668        taskQueues.addTaskToBackendTotalOrderQueue(killBlockingThreadTask); 2669      } 2670 2671      // Wait for thread termination 2672 for (Iterator iter = workerThreads.iterator(); iter.hasNext();) 2673      { 2674        BackendWorkerThread thread = (BackendWorkerThread) iter.next(); 2675        if (thread != Thread.currentThread()) 2676        { // Do not try to wait for self if we are the one who started the 2677 // disabling. 2678 try 2679          { 2680            thread.join(); 2681          } 2682          catch (InterruptedException ignore) 2683          { 2684          } 2685        } 2686      } 2687 2688      // Remove the threads from the list 2689 workerThreads.clear(); 2690 2691      // Cleanup what could remain in the queues 2692 taskQueues.abortRemainingRequests(); 2693    } 2694  } 2695 2696  /** 2697   * Terminate the Deadlock Detection Thread. Throws a RuntimeException is the 2698   * thread was already stopped (or not started). 2699   */ 2700  public void terminateDeadlockDetectionThread() 2701  { 2702    taskQueues.terminateDeadlockDetectionThread(); 2703  } 2704 2705  /** 2706   * Convert a <code>&lt;DatabaseBackend&gt;List</code> to a 2707   * <code>&lt;BackendInfo&gt;List</code>. 2708   * <p> 2709   * The DatabaseBackend objects cannot be serialized because they are used as 2710   * MBean and notification emitters, so we want to extract the BackendInfo 2711   * (which are <code>Serializable</code>) out of them. 2712   * </p> 2713   * <p> 2714   * <strong>This method does not keep the XML configuration of the BackendInfo 2715   * objects. Subsequent calls to getXml() on BackendInfos returned by this 2716   * method will always return <code>null</code> 2717   * </p> 2718   * 2719   * @param backends a <code>List</code> of <code>DatabaseBackends</code> 2720   * @return a <code>List</code> of <code>BackendInfo</code> (possibly empty 2721   * if the list of backends was <code>null</code> 2722   * @see BackendInfo#toDatabaseBackends(VirtualDatabase, List) 2723   */ 2724  public static/* <BackendInfo> */List toBackendInfos( 2725  /* <DatabaseBackend> */List backends) 2726  { 2727    if (backends == null) 2728    { 2729      return new ArrayList (); 2730    } 2731    List backendInfos = new ArrayList (backends.size()); 2732    for (Iterator iter = backends.iterator(); iter.hasNext();) 2733    { 2734      DatabaseBackend backend = (DatabaseBackend) iter.next(); 2735      BackendInfo backendInfo = new BackendInfo(backend); 2736      // we do not keep the XML configuration in the BackendInfo 2737 // FIXME I did that to mimic current behavior but I don't see the 2738 // reason why (maybe the size of XML to transfer on the wire?) 2739 backendInfo.setXml(null); 2740      backendInfos.add(backendInfo); 2741    } 2742    return backendInfos; 2743  } 2744 2745  /** 2746   * Check if the given vdb user is a valid user for this backend. 2747   * 2748   * @param vdbUser to be checked. 2749   * @return true if the vdb user is valid, false otherwise. 2750   */ 2751  public boolean isValidBackendUser(VirtualDatabaseUser vdbUser) 2752  { 2753    Connection conn = null; 2754    try 2755    { 2756      conn = DriverManager.getConnection(url, vdbUser.getLogin(), vdbUser 2757          .getPassword(), driverPath, driverClassName); 2758      return true; 2759    } 2760    catch (SQLException ignore) 2761    { 2762      if (logger.isDebugEnabled()) 2763      { 2764        logger.debug("Failed to get connection using vdb user " 2765            + vdbUser.getLogin() + " as real user", ignore); 2766      } 2767      return false; 2768    } 2769    finally 2770    { 2771      if (conn != null) 2772      { 2773        try 2774        { 2775          conn.close(); 2776        } 2777        catch (SQLException ignore) 2778        { 2779          // Silently ignore 2780 } 2781      } 2782    } 2783  } 2784 2785  /** 2786   * Get the log4j logger for this backend 2787   * 2788   * @return the logger for this backend 2789   */ 2790  public Trace getLogger() 2791  { 2792    return logger; 2793  } 2794}

A to Z: JavaDoc & Examples Daily Java News & Articles Open Source Projects Open Source Codes Free Computer Books Remove Frame Popular Tags