AbstractLoadBalancer.java

/**
 * C-JDBC: Clustered JDBC.
 * Copyright (C) 2002-2004 French National Institute For Research In Computer
 * Science And Control (INRIA).
 * Contact: c-jdbc@objectweb.org
 * 
 * This library is free software; you can redistribute it and/or modify it
 * under the terms of the GNU Lesser General Public License as published by the
 * Free Software Foundation; either version 2.1 of the License, or any later
 * version.
 * 
 * This library is distributed in the hope that it will be useful, but WITHOUT
 * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
 * FITNESS FOR A PARTICULAR PURPOSE. See the GNU Lesser General Public License
 * for more details.
 * 
 * You should have received a copy of the GNU Lesser General Public License
 * along with this library; if not, write to the Free Software Foundation,
 * Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA.
 *
 * Initial developer(s): Emmanuel Cecchet.
 * Contributor(s): Vadim Kassin, Jaco Swart
 */

package org.objectweb.cjdbc.controller.loadbalancer;

import java.sql.CallableStatement;
import java.sql.Connection;
import java.sql.PreparedStatement;
import java.sql.SQLException;
import java.sql.Statement;
import java.util.ArrayList;

import javax.management.NotCompliantMBeanException;

import org.objectweb.cjdbc.common.exceptions.BadConnectionException;
import org.objectweb.cjdbc.common.exceptions.UnreachableBackendException;
import org.objectweb.cjdbc.common.i18n.Translate;
import org.objectweb.cjdbc.common.jmx.mbeans.AbstractLoadBalancerMBean;
import org.objectweb.cjdbc.common.log.Trace;
import org.objectweb.cjdbc.common.sql.AbstractRequest;
import org.objectweb.cjdbc.common.sql.AbstractWriteRequest;
import org.objectweb.cjdbc.common.sql.SelectRequest;
import org.objectweb.cjdbc.common.sql.StoredProcedure;
import org.objectweb.cjdbc.common.sql.filters.MacrosHandler;
import org.objectweb.cjdbc.common.xml.DatabasesXmlTags;
import org.objectweb.cjdbc.common.xml.XmlComponent;
import org.objectweb.cjdbc.controller.backend.DatabaseBackend;
import org.objectweb.cjdbc.controller.backend.DriverCompliance;
import org.objectweb.cjdbc.controller.cache.metadata.MetadataCache;
import org.objectweb.cjdbc.controller.connection.AbstractConnectionManager;
import org.objectweb.cjdbc.controller.jmx.AbstractStandardMBean;
import org.objectweb.cjdbc.controller.requestmanager.TransactionMarkerMetaData;
import org.objectweb.cjdbc.controller.virtualdatabase.ControllerResultSet;
import org.objectweb.cjdbc.controller.virtualdatabase.VirtualDatabase;

/**
 * The Request Load Balancer should implement the load balancing of the requests
 * among the backend nodes.
 * <p>
 * The requests comes from the Request Controller and are sent to the Connection
 * Managers.
 * 
 * @author <a href="mailto:Emmanuel.Cecchet@inria.fr">Emmanuel Cecchet </a>
 * @author <a href="mailto:vadim@kase.kz">Vadim Kassin </a>
 * @author <a href="mailto:jaco.swart@iblocks.co.uk">Jaco Swart </a>
 * @version 1.0
 */
public abstract class AbstractLoadBalancer extends AbstractStandardMBean
    implements
      XmlComponent,
      AbstractLoadBalancerMBean
{

  //
  // How the code is organized ?
  //
  // 1. Member variables/Constructor
  // 2. Getter/Setter (possibly in alphabetical order)
  // 3. Request handling
  // 4. Transaction management
  // 5. Backend management
  // 6. Debug/Monitoring
  //

  // Virtual Database this load balancer is attached to.
  protected VirtualDatabase vdb;
  protected int             raidbLevel;
  protected int             parsingGranularity;

  protected static Trace    logger = Trace
                                       .getLogger("org.objectweb.cjdbc.controller.loadbalancer");

  protected MacrosHandler   macroHandler;

  /**
   * Generic constructor that sets some member variables and checks that
   * backends are in the disabled state
   * 
   * @param vdb The virtual database this load balancer belongs to
   * @param raidbLevel The RAIDb level of this load balancer
   * @param parsingGranularity The parsing granularity needed by this load
   *          balancer
   */
  protected AbstractLoadBalancer(VirtualDatabase vdb, int raidbLevel,
      int parsingGranularity) throws SQLException, NotCompliantMBeanException
  {
    super(AbstractLoadBalancerMBean.class);
    this.raidbLevel = raidbLevel;
    this.parsingGranularity = parsingGranularity;
    this.vdb = vdb;
    try
    {
      vdb.acquireReadLockBackendLists();
    }
    catch (InterruptedException e)
    {
      String msg = Translate.get(
          "loadbalancer.backendlist.acquire.readlock.failed", e);
      logger.error(msg);
      throw new SQLException(msg);
    }
    int size = vdb.getBackends().size();
    ArrayList backends = vdb.getBackends();
    for (int i = 0; i < size; i++)
    {
      DatabaseBackend backend = (DatabaseBackend) backends.get(i);
      if (backend.isReadEnabled() || backend.isWriteEnabled())
      {
        if (logger.isWarnEnabled())
          logger.warn(Translate.get(
              "loadbalancer.constructor.backends.not.disabled", backend
                  .getName()));
        try
        {
          disableBackend(backend);
        }
        catch (Exception e)
        { // Set the disabled state anyway
          backend.disable();
        }
      }
    }
    vdb.releaseReadLockBackendLists();
  }

  //
  // Getter/Setter methods
  //

  /**
   * Returns the RAIDbLevel.
   * 
   * @return int the RAIDb level
   */
  public int getRAIDbLevel()
  {
    return raidbLevel;
  }

  /**
   * Sets the RAIDbLevel.
   * 
   * @param raidbLevel The RAIDb level to set
   */
  public void setRAIDbLevel(int raidbLevel)
  {
    this.raidbLevel = raidbLevel;
  }

  /**
   * Get the needed query parsing granularity.
   * 
   * @return needed query parsing granularity
   */
  public int getParsingGranularity()
  {
    return parsingGranularity;
  }

  /**
   * Set the needed query parsing granularity.
   * 
   * @param parsingGranularity the granularity to set
   */
  public void setParsingGranularity(int parsingGranularity)
  {
    this.parsingGranularity = parsingGranularity;
  }

  //
  // Request Handling
  //

  /**
   * Perform a read request. It is up to the implementation to choose to which
   * backend node(s) this request should be sent.
   * 
   * @param request an <code>SelectRequest</code>
   * @param metadataCache MetadataCache (null if none)
   * @return the corresponding <code>ControllerResultSet</code>
   * @exception SQLException if an error occurs
   */
  public abstract ControllerResultSet execReadRequest(SelectRequest request,
      MetadataCache metadataCache) throws SQLException;

  /**
   * Perform a write request. This request should usually be broadcasted to all
   * nodes.
   * 
   * @param request an <code>AbstractWriteRequest</code>
   * @return number of rows affected by the request
   * @throws AllBackendsFailedException if all backends failed to execute the
   *           request
   * @exception SQLException if an error occurs
   */
  public abstract int execWriteRequest(AbstractWriteRequest request)
      throws AllBackendsFailedException, SQLException;

  /**
   * Perform a write request and return a ResultSet containing the auto
   * generated keys.
   * 
   * @param request an <code>AbstractWriteRequest</code>
   * @param metadataCache MetadataCache (null if none)
   * @return auto generated keys
   * @throws AllBackendsFailedException if all backends failed to execute the
   *           request
   * @exception SQLException if an error occurs
   */
  public abstract ControllerResultSet execWriteRequestWithKeys(
      AbstractWriteRequest request, MetadataCache metadataCache)
      throws AllBackendsFailedException, SQLException;

  /**
   * Call a read-only stored procedure that returns a ResultSet. The stored
   * procedure will be executed by one node only.
   * 
   * @param proc the stored procedure call
   * @param metadataCache MetadataCache (null if none)
   * @return a <code>ControllerResultSet</code> value
   * @exception SQLException if an error occurs
   */
  public abstract ControllerResultSet execReadOnlyReadStoredProcedure(
      StoredProcedure proc, MetadataCache metadataCache) throws SQLException;

  /**
   * Call a stored procedure that returns a ResultSet. This stored procedure can
   * possibly perform writes and will therefore be executed by all nodes.
   * 
   * @param proc the stored procedure call
   * @param metadataCache MetadataCache (null if none)
   * @return a <code>ControllerResultSet</code> value
   * @exception SQLException if an error occurs
   */
  public abstract ControllerResultSet execReadStoredProcedure(
      StoredProcedure proc, MetadataCache metadataCache) throws SQLException;

  /**
   * Call a stored procedure that performs an update.
   * 
   * @param proc the stored procedure call
   * @return number of rows affected
   * @exception SQLException if an error occurs
   */
  public abstract int execWriteStoredProcedure(StoredProcedure proc)
      throws SQLException;

  /**
   * Execute a statement on a backend. If the execution fails, the connection is
   * checked for validity. If the connection was not valid, the query is
   * automatically retried on a new connection.
   * 
   * @param request the request to execute
   * @param backend the backend on which the request is executed
   * @param c connection used to create the statement
   * @param metadataCache MetadataCache (null if none)
   * @return the ControllerResultSet
   * @throws SQLException if an error occurs
   * @throws BadConnectionException if the connection was bad
   */
  public static final ControllerResultSet executeSelectRequestOnBackend(
      SelectRequest request, DatabaseBackend backend, Connection c,
      MetadataCache metadataCache) throws SQLException, BadConnectionException
  {
    ControllerResultSet rs = null;
    try
    {
      backend.addPendingReadRequest(request);
      String sql = request.getSQL();
      // Rewrite the query if needed
      sql = backend.rewriteQuery(sql);

      Statement s; // Can also be used as a PreparedStatement
      if (request.isDriverProcessed() || (request.getSqlSkeleton() == null))
        s = c.createStatement();
      else
      {
        s = c.prepareStatement(request.getSqlSkeleton());
        org.objectweb.cjdbc.driver.PreparedStatement.setPreparedStatement(sql,
            (PreparedStatement) s);
      }

      // Execute the query
      DriverCompliance driverCompliance = backend.getDriverCompliance();
      if (driverCompliance.supportSetQueryTimeout())
        s.setQueryTimeout(request.getTimeout());
      if ((request.getCursorName() != null)
          && (driverCompliance.supportSetCursorName()))
        s.setCursorName(request.getCursorName());
      if ((request.getFetchSize() != 0)
          && driverCompliance.supportSetFetchSize())
        s.setFetchSize(request.getFetchSize());
      if ((request.getMaxRows() > 0) && driverCompliance.supportSetMaxRows())
        s.setMaxRows(request.getMaxRows());
      if (request.isDriverProcessed() || (request.getSqlSkeleton() == null))
        rs = new ControllerResultSet(request, s.executeQuery(sql),
            metadataCache, s);
      else
        rs = new ControllerResultSet(request, ((PreparedStatement) s)
            .executeQuery(), metadataCache, s);
    }
    catch (SQLException e)
    { // Something bad happened
      if (backend.isValidConnection(c))
        throw e; // Connection is valid, throw the exception
      else
        throw new BadConnectionException(e);
    }
    finally
    {
      backend.removePendingRequest(request);
    }
    return rs;
  }

  /**
   * Execute an update prepared statement on a backend. If the execution fails,
   * the connection is checked for validity. If the connection was not valid,
   * the query is automatically retried on a new connection.
   * 
   * @param request the request to execute
   * @param backend the backend on which the request is executed
   * @param c connection used to create the statement
   * @return int Number of rows effected
   * @throws SQLException if an error occurs
   * @throws BadConnectionException if the connection was bad
   */
  public static final int executeUpdateRequestOnBackend(
      AbstractWriteRequest request, DatabaseBackend backend, Connection c)
      throws SQLException, BadConnectionException
  {
    try
    {
      backend.addPendingWriteRequest(request);
      String sql = request.getSQL();
      // Rewrite the query if needed
      sql = backend.rewriteQuery(sql);

      Statement s; // Can also be used as a PreparedStatement
      if (request.isDriverProcessed() || (request.getSqlSkeleton() == null))
        s = c.createStatement();
      else
      {
        s = c.prepareStatement(request.getSqlSkeleton());
        org.objectweb.cjdbc.driver.PreparedStatement.setPreparedStatement(sql,
            (PreparedStatement) s);
      }

      // Execute the query
      DriverCompliance driverCompliance = backend.getDriverCompliance();
      if (driverCompliance.supportSetQueryTimeout())
        s.setQueryTimeout(request.getTimeout());

      int rows = 0;
      if (request.isDriverProcessed() || (request.getSqlSkeleton() == null))
        rows = s.executeUpdate(sql);
      else
        rows = ((PreparedStatement) s).executeUpdate();

      s.close();
      return rows;
    }
    catch (SQLException e)
    { // Something bad happened
      if (backend.isValidConnection(c))
        throw e; // Connection is valid, throw the exception
      else
        throw new BadConnectionException(e);
    }
    finally
    {
      backend.removePendingRequest(request);
    }
  }

  /**
   * Execute an update prepared statement on a backend. If the execution fails,
   * the connection is checked for validity. If the connection was not valid,
   * the query is automatically retried on a new connection.
   * 
   * @param request the request to execute
   * @param backend the backend on which the request is executed
   * @param c connection used to create the statement
   * @param metadataCache MetadataCache (null if none)
   * @return ControllerResultSet containing the auto-generated keys
   * @throws SQLException if an error occurs
   * @throws BadConnectionException if the connection was bad
   */
  public static final ControllerResultSet executeUpdateRequestOnBackendWithKeys(
      AbstractWriteRequest request, DatabaseBackend backend, Connection c,
      MetadataCache metadataCache) throws SQLException, BadConnectionException
  {
    try
    {
      backend.addPendingWriteRequest(request);
      String sql = request.getSQL();
      // Rewrite the query if needed
      sql = backend.rewriteQuery(sql);

      Statement s = c.createStatement();

      // Execute the query
      DriverCompliance driverCompliance = backend.getDriverCompliance();
      if (driverCompliance.supportSetQueryTimeout())
        s.setQueryTimeout(request.getTimeout());

      s.executeUpdate(sql, Statement.RETURN_GENERATED_KEYS);
      ControllerResultSet rs = new ControllerResultSet(request, s
          .getGeneratedKeys(), metadataCache, s);
      return rs;
    }
    catch (SQLException e)
    { // Something bad happened
      if (backend.isValidConnection(c))
        throw e; // Connection is valid, throw the exception
      else
        throw new BadConnectionException(e);
    }
    finally
    {
      backend.removePendingRequest(request);
    }
  }

  /**
   * Execute a read stored procedure on the given backend. The callable
   * statement is setXXX if the driver has not processed the statement.
   * 
   * @param proc the stored procedure to execute
   * @param backend the backend on which to execute the stored procedure
   * @param c the connection on which to execute the stored procedure
   * @param metadataCache the matedatacache to build the ControllerResultSet
   * @return the controllerResultSet
   * @throws SQLException if an error occurs
   * @throws BadConnectionException if the connection was bad
   */
  public static final ControllerResultSet executeReadStoredProcedureOnBackend(
      StoredProcedure proc, DatabaseBackend backend, Connection c,
      MetadataCache metadataCache) throws SQLException, BadConnectionException
  {
    try
    {
      backend.addPendingReadRequest(proc);

      // We suppose here that the request does not modify the schema since
      // it is a read-only query.
      CallableStatement cs;
      if (proc.isDriverProcessed())
        cs = c.prepareCall(proc.getSQL());
      else
      {
        cs = c.prepareCall(proc.getSqlSkeleton());
        org.objectweb.cjdbc.driver.PreparedStatement.setPreparedStatement(proc
            .getSQL(), cs);
      }
      if (backend.getDriverCompliance().supportSetQueryTimeout())
        cs.setQueryTimeout(proc.getTimeout());
      if ((proc.getMaxRows() > 0)
          && backend.getDriverCompliance().supportSetMaxRows())
        cs.setMaxRows(proc.getMaxRows());
      ControllerResultSet rs = new ControllerResultSet(proc, cs.executeQuery(),
          metadataCache, cs);
      return rs;
    }
    catch (SQLException e)
    { // Something bad happened
      if (backend.isValidConnection(c))
        throw e; // Connection is valid, throw the exception
      else
        throw new BadConnectionException(e);
    }
    finally
    {
      backend.removePendingRequest(proc);
    }
  }

  /**
   * Execute a write stored procedure on the given backend. The callable
   * statement is setXXX if the driver has not processed the statement.
   * 
   * @param proc the stored procedure to execute
   * @param backend the backend on which to execute the stored procedure
   * @param c the connection on which to execute the stored procedure
   * @return the number of updated rows
   * @throws SQLException if an error occurs
   * @throws BadConnectionException if the connection was bad
   */
  public static final int executeWriteStoredProcedureOnBackend(
      StoredProcedure proc, DatabaseBackend backend, Connection c)
      throws SQLException, BadConnectionException
  {
    try
    {
      backend.addPendingWriteRequest(proc);

      // We suppose here that the request does not modify the schema since
      // it is a read-only query.
      CallableStatement cs;
      if (proc.isDriverProcessed())
        cs = c.prepareCall(proc.getSQL());
      else
      {
        cs = c.prepareCall(proc.getSqlSkeleton());
        org.objectweb.cjdbc.driver.PreparedStatement.setPreparedStatement(proc
            .getSQL(), cs);
      }
      if (backend.getDriverCompliance().supportSetQueryTimeout())
        cs.setQueryTimeout(proc.getTimeout());
      if ((proc.getMaxRows() > 0)
          && backend.getDriverCompliance().supportSetMaxRows())
        cs.setMaxRows(proc.getMaxRows());
      int rows = cs.executeUpdate();
      cs.close();
      return rows;
    }
    catch (SQLException e)
    { // Something bad happened
      if (backend.isValidConnection(c))
        throw e; // Connection is valid, throw the exception
      else
        throw new BadConnectionException(e);
    }
    finally
    {
      backend.removePendingRequest(proc);
    }
  }

  //
  // Transaction management
  //

  /**
   * Begin a new transaction.
   * 
   * @param tm The transaction marker metadata
   * @throws SQLException if an error occurs
   */
  public abstract void begin(TransactionMarkerMetaData tm) throws SQLException;

  /**
   * Commit a transaction.
   * 
   * @param tm The transaction marker metadata
   * @throws AllBackendsFailedException if all backends failed to execute the
   *           request
   * @exception SQLException if an error occurs
   */
  public abstract void commit(TransactionMarkerMetaData tm)
      throws AllBackendsFailedException, SQLException;

  /**
   * Rollback a transaction.
   * 
   * @param tm The transaction marker metadata
   * @throws AllBackendsFailedException if all backends failed to execute the
   *           request
   * @exception SQLException if an error occurs
   */
  public abstract void rollback(TransactionMarkerMetaData tm)
      throws AllBackendsFailedException, SQLException;

  //
  // Backends management
  //

  /**
   * Enable a backend without further check. The backend is at least read
   * enabled but could also be enabled for writes. Ask the corresponding
   * connection manager to initialize the connections if needed.
   * 
   * @param db The database backend to enable
   * @param writeEnabled True if the backend must be enabled for writes
   * @throws SQLException if an error occurs
   */
  public abstract void enableBackend(DatabaseBackend db, boolean writeEnabled)
      throws SQLException;

  /**
   * Disable a backend without further check. Ask the corresponding connection
   * manager to finalize the connections if needed. This method should not be
   * called directly but instead should access the
   * <code>RequestManager.disableBackeknd(...)</code> method.
   * 
   * @param db The database backend to disable
   * @throws SQLException if an error occurs
   */
  public abstract void disableBackend(DatabaseBackend db) throws SQLException;

  /**
   * Associate a weight to a backend identified by its logical name.
   * 
   * @param name the backend name
   * @param w the weight
   * @throws SQLException if an error occurs
   */
  public void setWeight(String name, int w) throws SQLException
  {
    throw new SQLException("Weight is not supported by this load balancer");
  }

  //
  // Debug/Monitoring
  //

  /**
   * Get information about the Request Load Balancer
   * 
   * @return <code>String</code> containing information
   */
  public abstract String getInformation();

  /**
   * Get information about the Request Load Balancer in xml
   * 
   * @return <code>String</code> containing information, xml formatted
   */
  public abstract String getXmlImpl();

  /**
   * This sets the macro handler for this load balancer. Handling macros
   * prevents different backends to generate different values when interpreting
   * the macros which could result in data inconsitencies.
   * 
   * @param handler <code>MacrosHandler</code> instance
   */
  public void setMacroHandler(MacrosHandler handler)
  {
    this.macroHandler = handler;
  }

  /**
   * Interprets the macros in the request (depending on the
   * <code>MacroHandler</code> set for this class) and modify either the
   * skeleton or the query itself. Note that the given object is directly
   * modified.
   * 
   * @param request the request to process
   */
  public void handleMacros(AbstractRequest request)
  {
    if (macroHandler == null)
      return;

    if (request.isWriteRequest())
    {
      AbstractWriteRequest writeReq = (AbstractWriteRequest) request;
      if (writeReq.isCreate() || writeReq.isAlter() || writeReq.isDrop())
        return;
    }

    if (request.isDriverProcessed() || (request.getSqlSkeleton() == null))
      request.setSQL(macroHandler.processMacros(request.getSQL()));
    else
      request.setSqlSkeleton(macroHandler.processMacros(request
          .getSqlSkeleton()));
  }

  /**
   * @see org.objectweb.cjdbc.common.xml.XmlComponent#getXml()
   */
  public String getXml()
  {
    StringBuffer info = new StringBuffer();
    info.append("<" + DatabasesXmlTags.ELT_LoadBalancer + ">");
    info.append(getXmlImpl());
    info.append("</" + DatabasesXmlTags.ELT_LoadBalancer + ">");
    return info.toString();
  }

  /**
   * Factorized code to start a transaction on a backend and to retrieve a
   * connection on this backend
   * 
   * @param backend the backend needed to check valid connection against this
   *          backend test statement
   * @param cm the connection manager to use to retrieve connections
   * @param tid the id of the transaction to start
   * @return a valid connection with a started transaction
   * @throws SQLException if the backend is valid but set autocommit cannot be
   *           set to false
   * @throws UnreachableBackendException if the backend is not reachable, ie not
   *           valid connection can be retrieved
   */
  public static final Connection getConnectionAndBeginTransaction(
      DatabaseBackend backend, AbstractConnectionManager cm, long tid)
      throws SQLException, UnreachableBackendException
  {
    Connection c = null;
    boolean isConnectionValid = false;
    do
    {
      c = cm.getConnection(tid);

      // Sanity check
      if (c == null)
        throw new UnreachableBackendException(Translate.get(
            "loadbalancer.unable.get.connection", new String[]{
                String.valueOf(tid), backend.getName()}));
      try
      {
        c.setAutoCommit(false);
        isConnectionValid = true;
      }
      catch (SQLException e)
      {
        if (backend.isValidConnection(c))
          throw e; // Connection is valid, throw the exception
        else
        {
          cm.deleteConnection(tid);
        }
      }
    }
    while (!isConnectionValid);
    return c;
  }

  /**
   * @see org.objectweb.cjdbc.controller.jmx.AbstractStandardMBean#getAssociatedString()
   */
  public String getAssociatedString()
  {
    return "loadbalancer";
  }
}

---