src/org/objectweb/cjdbc/controller/requestmanager/RequestManager.java

説明を見る。

package org.objectweb.cjdbc.controller.requestmanager;

import java.sql.SQLException;
import java.util.ArrayList;
import java.util.Hashtable;

import org.objectweb.cjdbc.common.exceptions.BackupException;
import org.objectweb.cjdbc.common.exceptions.ExceptionTypes;
import org.objectweb.cjdbc.common.exceptions.NoMoreBackendException;
import org.objectweb.cjdbc.common.exceptions.OctopusException;
import org.objectweb.cjdbc.common.exceptions.RollbackException;
import org.objectweb.cjdbc.common.i18n.Translate;
import org.objectweb.cjdbc.common.log.Trace;
import org.objectweb.cjdbc.common.shared.BackendState;
import org.objectweb.cjdbc.common.shared.BackupListener;
import org.objectweb.cjdbc.common.sql.AbstractWriteRequest;
import org.objectweb.cjdbc.common.sql.AlterRequest;
import org.objectweb.cjdbc.common.sql.CreateRequest;
import org.objectweb.cjdbc.common.sql.ParsingGranularities;
import org.objectweb.cjdbc.common.sql.RequestType;
import org.objectweb.cjdbc.common.sql.SelectRequest;
import org.objectweb.cjdbc.common.sql.StoredProcedure;
import org.objectweb.cjdbc.common.sql.UpdateRequest;
import org.objectweb.cjdbc.common.sql.schema.DatabaseSchema;
import org.objectweb.cjdbc.common.sql.schema.DatabaseTable;
import org.objectweb.cjdbc.common.xml.DatabasesXmlTags;
import org.objectweb.cjdbc.common.xml.XmlComponent;
import org.objectweb.cjdbc.controller.backend.BackendStateListener;
import org.objectweb.cjdbc.controller.backend.DatabaseBackend;
import org.objectweb.cjdbc.controller.backup.BackupManager;
import org.objectweb.cjdbc.controller.cache.metadata.MetadataCache;
import org.objectweb.cjdbc.controller.cache.parsing.ParsingCache;
import org.objectweb.cjdbc.controller.cache.result.AbstractResultCache;
import org.objectweb.cjdbc.controller.cache.result.entries.CacheEntry;
import org.objectweb.cjdbc.controller.loadbalancer.AbstractLoadBalancer;
import org.objectweb.cjdbc.controller.loadbalancer.AllBackendsFailedException;
import org.objectweb.cjdbc.controller.recoverylog.AbstractRecoveryLog;
import org.objectweb.cjdbc.controller.recoverylog.BackendRecoveryInfo;
import org.objectweb.cjdbc.controller.recoverylog.JDBCRecoverThread;
import org.objectweb.cjdbc.controller.scheduler.AbstractScheduler;
import org.objectweb.cjdbc.controller.virtualdatabase.ControllerResultSet;
import org.objectweb.cjdbc.controller.virtualdatabase.VirtualDatabase;

public class RequestManager implements XmlComponent
{

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

  protected long                 beginTimeout;
  protected long                 commitTimeout;
  protected long                 rollbackTimeout;

  protected VirtualDatabase      vdb;

  protected AbstractScheduler    scheduler;

  protected AbstractResultCache  resultCache;

  protected AbstractLoadBalancer loadBalancer;

  protected AbstractRecoveryLog  recoveryLog;

  protected BackupManager        backupManager;

  // Database schema pointing to the virtual dabase schema
  protected DatabaseSchema       dbs;
  private boolean                schemaIsStatic                = false;
  private boolean                isCaseSensitiveParsing        = false;
  protected ParsingCache         parsingCache                  = null;
  private MetadataCache          metadataCache                 = null;

  // SQL queries parsing granularity according to Scheduler, ResultCache and
  // LoadBalancer required granularity
  protected int                  schedulerParsingranularity    = ParsingGranularities.NO_PARSING;
  private int                    cacheParsingranularity        = ParsingGranularities.NO_PARSING;
  private int                    loadBalancerParsingranularity = ParsingGranularities.NO_PARSING;
  protected int                  requiredGranularity           = ParsingGranularities.NO_PARSING;
  protected Hashtable            tidLoginTable;

  protected static Trace         logger                        = null;
  private BackendStateListener   backendStateListener;

  //
  // Constructors
  //

  public RequestManager(VirtualDatabase vdb, AbstractScheduler scheduler,
      AbstractResultCache cache, AbstractLoadBalancer loadBalancer,
      AbstractRecoveryLog recoveryLog, long beginTimeout, long commitTimeout,
      long rollbackTimeout) throws SQLException
  {
    assignAndCheckSchedulerLoadBalancerValidity(scheduler, loadBalancer);
    // requiredGranularity is the maximum of each component granularity
    this.resultCache = cache;
    this.vdb = vdb;
    if (resultCache != null)
    {
      cacheParsingranularity = cache.getParsingGranularity();
      if (cacheParsingranularity > requiredGranularity)
        requiredGranularity = cacheParsingranularity;
    }
    setRecoveryLog(recoveryLog);
    initRequestManagerVariables(vdb, beginTimeout, commitTimeout,
        rollbackTimeout);
    setBackendsLastKnownCheckpointFromRecoveryLog();
    logger.info(Translate.get("requestmanager.parsing.granularity",
        ParsingGranularities.getInformation(requiredGranularity)));
  }

  private void setBackendsLastKnownCheckpointFromRecoveryLog()
  {

    if (recoveryLog == null)
      return;
    String databaseName = vdb.getName();
    ArrayList backends = vdb.getBackends();
    int size = backends.size();
    DatabaseBackend backend;
    BackendRecoveryInfo info;
    for (int i = 0; i < size; i++)
    {
      backend = (DatabaseBackend) backends.get(i);
      try
      {
        info = recoveryLog.getBackendInfo(databaseName, backend.getName());
        backend.setLastKnownCheckpoint(info.getLastCheckpoint());
      }
      catch (SQLException e)
      {
        logger.error(Translate.get("requestmanager.checkpoint.not.found",
            backend.getName()), e);
      }
    }

  }

  private void assignAndCheckSchedulerLoadBalancerValidity(
      AbstractScheduler scheduler, AbstractLoadBalancer loadBalancer)
      throws SQLException
  {
    if (scheduler == null)
      throw new SQLException(Translate.get("requestmanager.null.scheduler"));

    if (loadBalancer == null)
      throw new SQLException(Translate.get("requestmanager.null.loadbalancer"));

    if (scheduler.getRAIDbLevel() != loadBalancer.getRAIDbLevel())
      throw new SQLException(Translate.get(
          "requestmanager.incompatible.raidb.levels",
          new String[]{"" + scheduler.getRAIDbLevel(),
              "" + loadBalancer.getRAIDbLevel()}));

    // requiredGranularity is the maximum of each component granularity
    this.scheduler = scheduler;
    schedulerParsingranularity = scheduler.getParsingGranularity();
    requiredGranularity = schedulerParsingranularity;
    this.loadBalancer = loadBalancer;
    loadBalancerParsingranularity = loadBalancer.getParsingGranularity();
    if (loadBalancerParsingranularity > requiredGranularity)
      requiredGranularity = loadBalancerParsingranularity;
  }

  private void initRequestManagerVariables(VirtualDatabase vdb,
      long beginTimeout, long commitTimeout, long rollbackTimeout)
  {
    this.tidLoginTable = new Hashtable();
    this.beginTimeout = beginTimeout;
    this.commitTimeout = commitTimeout;
    this.rollbackTimeout = rollbackTimeout;
    this.vdb = vdb;
    logger = Trace.getLogger("org.objectweb.cjdbc.controller.RequestManager."
        + vdb.getDatabaseName());
  }

  //
  // Request Handling
  //

  public ControllerResultSet execReadRequest(SelectRequest request)
      throws SQLException
  {
    // Sanity check
    if (!request.isAutoCommit())
    { // Check that the transaction has been started
      long tid = request.getTransactionId();
      if (!tidLoginTable.containsKey(new Long(tid)))
        throw new SQLException(Translate.get("transaction.not.started", tid));
    }

    // If we need to parse the request, try to get the parsing from the cache.
    // Note that if we have a cache miss but backgroundParsing has been turned
    // on, then this call will start a ParsedThread in background.
    if ((requiredGranularity != ParsingGranularities.NO_PARSING)
        && (!request.isParsed()))
    {
      if (parsingCache == null)
        request.parse(dbs, requiredGranularity, isCaseSensitiveParsing);
      else
        parsingCache.getParsingFromCache(request);
    }

    //
    // SCHEDULER
    //

    // Get the parsing now if the request is not yet parsed. The parsing is
    // handled by the ParsingCache that may already have parsed the request
    // in background (if backgroundParsing is set).
    if ((schedulerParsingranularity != ParsingGranularities.NO_PARSING)
        && !request.isParsed())
    {
      if (parsingCache == null)
        request.parse(dbs, requiredGranularity, isCaseSensitiveParsing);
      else
        parsingCache.getParsingFromCacheAndParseIfMissing(request);
    }

    // Wait for the scheduler to give us the authorization to execute
    scheduler.scheduleReadRequest(request);

    if (logger.isDebugEnabled())
      logger.debug(Translate.get("requestmanager.read.request", new String[]{
          request.getId() + "",
          request.getSQLShortForm(vdb.getSQLShortFormLength())}));

    //
    // CACHE
    //

    ControllerResultSet result = null;
    try
    { // Check cache if any
      if (resultCache != null)
      {
        CacheEntry qce = resultCache.getFromCache(request, true);
        if (qce != null)
        {
          result = qce.getResult();
          if (result != null)
          { // Cache hit !
            if (vdb.getSQLMonitor() != null)
              vdb.getSQLMonitor().logCacheHit(request);

            scheduler.readCompleted(request);
            return result;
          }
        }
      }

      //
      // LOAD BALANCER
      //

      // At this point, we have a result cache miss.
      // If we had a parsing cache miss too, wait for the parsing to be done if
      // needed.
      if ((loadBalancerParsingranularity != ParsingGranularities.NO_PARSING)
          && !request.isParsed())
      {
        if (parsingCache == null)
          request.parse(dbs, requiredGranularity, isCaseSensitiveParsing);
        else
          parsingCache.getParsingFromCacheAndParseIfMissing(request);
      }

      // Send the request to the load balancer
      result = loadBalancer.execReadRequest(request, metadataCache);

      //
      // UPDATES & NOTIFICATIONS
      //

      // Update cache
      if ((resultCache != null)
          && (request.getCacheAbility() != RequestType.UNCACHEABLE))
      {
        if (!request.isParsed()
            && (cacheParsingranularity != ParsingGranularities.NO_PARSING))
        { // The cache was the only one to need parsing and
          // the request was not previously in the cache
          {
            if (parsingCache == null)
              request.parse(dbs, requiredGranularity, isCaseSensitiveParsing);
            else
              parsingCache.getParsingFromCacheAndParseIfMissing(request);
          }
        }
        resultCache.addToCache(request, result);
      }
    }
    catch (Exception failed)
    {
      if (resultCache != null)
        resultCache.removeFromPendingQueries(request);
      scheduler.readCompleted(request);
      if (failed instanceof NoMoreBackendException)
        throw (NoMoreBackendException) failed;
      String msg = Translate.get("requestmanager.request.failed", new String[]{
          request.getSQLShortForm(vdb.getSQLShortFormLength()),
          failed.getMessage()});
      if (failed instanceof RuntimeException)
        logger.warn(msg, failed);
      else
        logger.warn(msg);
      throw new SQLException(msg);
    }

    // Notify scheduler of completion
    scheduler.readCompleted(request);

    return result;
  }

  public int execWriteRequest(AbstractWriteRequest request) throws SQLException
  {
    scheduleExecWriteRequest(request);
    int execWriteRequestResult = 0;
    try
    {
      execWriteRequestResult = loadBalanceExecWriteRequest(request);
    }
    catch (AllBackendsFailedException e)
    {
      logger.fatal(Translate
          .get("requestmanager.write.request.failed.unexpected"), e);
    }
    catch (NoMoreBackendException e)
    {
      logger.fatal(Translate
          .get("requestmanager.write.request.failed.no.more.backends"), e);
    }
    updateAndNotifyExecWriteRequest(request, true);
    return execWriteRequestResult;
  }

  public ControllerResultSet execWriteRequestWithKeys(
      AbstractWriteRequest request) throws SQLException
  {
    scheduleExecWriteRequest(request);
    ControllerResultSet execWriteRequestWithKeysResult = null;
    try
    {
      execWriteRequestWithKeysResult = loadBalanceExecWriteRequestWithKeys(request);
    }
    catch (AllBackendsFailedException e)
    {
      logger.fatal(Translate
          .get("requestmanager.write.request.keys.failed.unexpected"), e);
    }
    updateAndNotifyExecWriteRequest(request, true);
    return execWriteRequestWithKeysResult;
  }

  public void scheduleExecWriteRequest(AbstractWriteRequest request)
      throws SQLException
  {
    // Sanity check
    if (!request.isAutoCommit())
    { // Check that the transaction has been started
      long tid = request.getTransactionId();
      if (!tidLoginTable.containsKey(new Long(tid)))
        throw new SQLException(Translate.get("transaction.not.started", tid));
    }

    // If we need to parse the request, try to get the parsing from the cache.
    // Note that if we have a cache miss but backgroundParsing has been turned
    // on, then this call will start a ParsedThread in background.
    if ((requiredGranularity != ParsingGranularities.NO_PARSING)
        && (!request.isParsed()))
    {
      if (parsingCache == null)
        request.parse(dbs, requiredGranularity, isCaseSensitiveParsing);
      else
        parsingCache.getParsingFromCache(request);
    }

    //
    // SCHEDULER
    //

    // Get the parsing now if the request is not yet parsed. The parsing is
    // handled by the ParsingCache that may already have parsed the request
    // in background (if backgroundParsing is set).
    if ((schedulerParsingranularity != ParsingGranularities.NO_PARSING)
        && !request.isParsed())
    {
      if (parsingCache == null)
        request.parse(dbs, requiredGranularity, isCaseSensitiveParsing);
      else
        parsingCache.getParsingFromCacheAndParseIfMissing(request);
    }

    // Wait for the scheduler to give us the authorization to execute
    try
    {
      scheduler.scheduleWriteRequest(request);
    }
    catch (RollbackException e)
    { // Something bad happened and we need to rollback this transaction
      rollback(request.getTransactionId());
      throw new SQLException(e.getMessage());
    }

    if (logger.isDebugEnabled())
      logger.debug(Translate.get("requestmanager.write.request", new String[]{
          String.valueOf(request.getId()),
          request.getSQLShortForm(vdb.getSQLShortFormLength())}));

    // If we have a parsing cache miss, wait for the parsing to be done if
    // needed. Note that even if the cache was the only one to require parsing,
    // we wait for the parsing result here, because if it fails, we must not
    // execute the query.
    try
    {
      if ((requiredGranularity != ParsingGranularities.NO_PARSING)
          && !request.isParsed())
      {
        if (parsingCache == null)
          request.parse(dbs, requiredGranularity, isCaseSensitiveParsing);
        else
          parsingCache.getParsingFromCacheAndParseIfMissing(request);
      }
    }
    catch (SQLException e)
    {
      // If the parsing fail, we must release the lock acquired ...
      scheduler.writeCompleted(request);
      throw e;
    }
  }

  public ControllerResultSet loadBalanceExecWriteRequestWithKeys(
      AbstractWriteRequest request) throws AllBackendsFailedException,
      SQLException
  {
    try
    { // Send the request to the load balancer
      return loadBalancer.execWriteRequestWithKeys(request, metadataCache);
    }
    catch (Exception failed)
    {
      scheduler.writeCompleted(request);
      String msg = Translate.get("requestmanager.request.failed", new String[]{
          request.getSQLShortForm(vdb.getSQLShortFormLength()),
          failed.getMessage()});
      if (failed instanceof RuntimeException)
        logger.warn(msg, failed);
      else
        logger.warn(msg);
      if (failed instanceof AllBackendsFailedException)
        throw (AllBackendsFailedException) failed;
      else
        throw new SQLException(msg);
    }
  }

  public int loadBalanceExecWriteRequest(AbstractWriteRequest request)
      throws AllBackendsFailedException, SQLException
  {
    try
    { // Send the request to the load balancer
      if (request.isUpdate() && (resultCache != null))
      { // Try the optimization if we try to update values that are already
        // up-to-date.
        if (!resultCache.isUpdateNecessary((UpdateRequest) request))
          return 0;
      }
      return loadBalancer.execWriteRequest(request);
    }
    catch (Exception failed)
    {
      scheduler.writeCompleted(request);
      String msg = Translate.get("requestmanager.request.failed", new String[]{
          request.getSQLShortForm(vdb.getSQLShortFormLength()),
          failed.getMessage()});
      if (failed instanceof RuntimeException)
        logger.warn(msg, failed);
      else
        logger.warn(msg);
      if (failed instanceof AllBackendsFailedException)
        throw (AllBackendsFailedException) failed;
      else
        throw new SQLException(msg);
    }
  }

  public void updateAndNotifyExecWriteRequest(AbstractWriteRequest request,
      boolean notifyScheduler) throws SQLException
  {

    try
    { // Notify cache if any
      if (resultCache != null)
      { // Update cache
        resultCache.writeNotify(request);
      }

      // Log the request
      if (recoveryLog != null)
        recoveryLog.logRequest(request);

      // Update the schema if needed
      if (requiredGranularity != ParsingGranularities.NO_PARSING)
      {
        if (request.isCreate())
        { // Add the table to the schema
          dbs.addTable(((CreateRequest) request).getDatabaseTable());
          if (logger.isDebugEnabled())
            logger.debug(Translate.get("requestmanager.schema.add.table",
                request.getTableName()));
        }
        else if (request.isDrop())
        { // Delete the table from the schema
          dbs.removeTable(dbs.getTable(request.getTableName()));
          if (logger.isDebugEnabled())
            logger.debug(Translate.get("requestmanager.schema.remove.table",
                request.getTableName()));
        }
        else if (request.isAlter()
            && (requiredGranularity > ParsingGranularities.TABLE))
        {
          // Add or drop the column from the table
          AlterRequest req = (AlterRequest) request;
          if (req.isDrop())
            dbs.getTable(req.getTableName()).remove(req.getColumn().getName());
          else if (req.isAdd())
          {
            dbs.getTable(req.getTableName()).addColumn(req.getColumn());
          }
        }
      }

      // Notify scheduler
      if (notifyScheduler)
        scheduler.writeCompleted(request);

    }
    catch (Exception failed)
    {
      scheduler.writeCompleted(request);
      String msg = Translate.get("requestmanager.request.failed", new String[]{
          request.getSQLShortForm(vdb.getSQLShortFormLength()),
          failed.getMessage()});
      if (failed instanceof RuntimeException)
        logger.warn(msg, failed);
      else
        logger.warn(msg);
      throw new SQLException(msg);
    }
  }

  public ControllerResultSet execReadStoredProcedure(StoredProcedure proc)
      throws SQLException
  {
    // Sanity check
    if (!proc.isAutoCommit())
    { // Check that the transaction has been started
      long tid = proc.getTransactionId();
      if (!tidLoginTable.containsKey(new Long(tid)))
        throw new SQLException(Translate.get("transaction.not.started", tid));
    }

    //
    // SCHEDULER
    //

    // Note that no parsing is required for stored procedures.
    // We build a fake request that span over all tables to be sure
    // that the scheduler will lock everything.
    SelectRequest spanOnAllTables = null;
    if (requiredGranularity != ParsingGranularities.NO_PARSING)
    {
      String sql = "SELECT * FROM";
      ArrayList tables = dbs.getTables();
      int size = tables.size();
      // The following code does not generate a syntaxically correct SQL
      // request (an extra coma at the beginning) but that is not an issue.
      for (int i = 0; i < size; i++)
        sql += " ," + ((DatabaseTable) tables.get(i)).getName();

      spanOnAllTables = new SelectRequest(sql, false, 0, proc
          .getLineSeparator());
    }
    else
      spanOnAllTables = new SelectRequest("select * from x", false, 0, proc
          .getLineSeparator());

    // Wait for the scheduler to give us the authorization to execute
    scheduler.scheduleReadRequest(spanOnAllTables);

    if (logger.isDebugEnabled())
      logger.debug(Translate.get("requestmanager.read.store.procedure",
          new String[]{String.valueOf(proc.getId()),
              proc.getSQLShortForm(vdb.getSQLShortFormLength())}));

    //
    // CACHE
    //

    // Cache is always flushed unless the user has explicitely set the
    // connection to read-only mode in which case we assume that the
    // users deliberately forces the cache not to be flushed when calling
    // this stored procedure.
    if ((resultCache != null) && (!proc.isReadOnly()))
      resultCache.flushCache();

    ControllerResultSet result = null;
    try
    {
      //
      // LOAD BALANCER
      //

      // Send the request to the load balancer
      if (proc.isReadOnly())
        result = loadBalancer.execReadOnlyReadStoredProcedure(proc,
            metadataCache);
      else
        result = loadBalancer.execReadStoredProcedure(proc, metadataCache);

      //
      // RECOVERY LOG
      //

      if (recoveryLog != null)
        recoveryLog.logRequest(proc, true);

    }
    catch (Exception failed)
    {
      scheduler.readCompleted(spanOnAllTables);
      String msg = Translate.get("requestmanager.store.procedure.failed",
          new String[]{proc.getSQLShortForm(vdb.getSQLShortFormLength()),
              failed.getMessage()});
      logger.warn(msg);
      throw new SQLException(msg);
    }

    // Notify scheduler of completion
    scheduler.readCompleted(spanOnAllTables);

    return result;
  }

  public int execWriteStoredProcedure(StoredProcedure proc) throws SQLException
  {
    // Sanity check
    if (!proc.isAutoCommit())
    { // Check that the transaction has been started
      long tid = proc.getTransactionId();
      if (!tidLoginTable.containsKey(new Long(tid)))
        throw new SQLException(Translate.get("transaction.not.started", String
            .valueOf(tid)));
    }

    //
    // SCHEDULER
    //

    // Note that no parsing is required for stored procedures.
    // We build a fake request that span over all tables to be sure
    // that the scheduler will lock everything.
    SelectRequest spanOnAllTables = null;
    if (requiredGranularity != ParsingGranularities.NO_PARSING)
    {
      String sql = "SELECT * FROM";
      ArrayList tables = dbs.getTables();
      int size = tables.size();
      // The following code does not generate a syntaxically correct SQL
      // request (an extra coma at the beginning) but that is not an issue.
      for (int i = 0; i < size; i++)
        sql += " ," + ((DatabaseTable) tables.get(i)).getName();

      spanOnAllTables = new SelectRequest(sql, false, 0, proc
          .getLineSeparator());
    }
    else
      spanOnAllTables = new SelectRequest("select * from x", false, 0, proc
          .getLineSeparator());

    // Wait for the scheduler to give us the authorization to execute
    // We schedule a read even if it is a write because there is no
    // way to lock all tables for write in a single query. FIXME!
    scheduler.scheduleReadRequest(spanOnAllTables);

    if (logger.isDebugEnabled())
      logger.debug(Translate.get("requestmanager.write.store.procedure",
          new String[]{String.valueOf(proc.getId()),
              proc.getSQLShortForm(vdb.getSQLShortFormLength())}));

    //
    // CACHE
    //

    // Flush cache if any
    if (resultCache != null)
      resultCache.flushCache();

    int result;
    try
    {
      //
      // LOAD BALANCER
      //

      // Send the request to the load balancer
      result = loadBalancer.execWriteStoredProcedure(proc);

      //
      // RECOVERY LOG
      //

      if (recoveryLog != null)
        recoveryLog.logRequest(proc, false);

    }
    catch (Exception failed)
    {
      scheduler.readCompleted(spanOnAllTables);
      String msg = Translate.get("requestmanager.store.procedure.failed",
          new String[]{proc.getSQLShortForm(vdb.getSQLShortFormLength()),
              failed.getMessage()});
      logger.warn(msg);
      throw new SQLException(msg);
    }

    // Notify scheduler of completion
    scheduler.readCompleted(spanOnAllTables);

    return result;
  }

  //
  // Transaction management
  //

  public long begin(String login) throws SQLException
  {
    try
    {
      TransactionMarkerMetaData tm = new TransactionMarkerMetaData(0,
          beginTimeout, login);

      // Wait for the scheduler to give us the authorization to execute
      long tid = scheduler.begin(tm);
      tm.setTransactionId(tid);

      if (logger.isDebugEnabled())
        logger.debug(Translate.get("transaction.begin", String.valueOf(tid)));

      try
      {
        // Send to load balancer
        loadBalancer.begin(tm);

        // Log the begin
        if (recoveryLog != null)
        {
          recoveryLog.begin(tm);
        }
      }
      catch (SQLException e)
      {
        throw e;
      }
      finally
      {
        // Notify scheduler for completion in any case
        scheduler.beginCompleted(tid);
      }

      tidLoginTable.put(new Long(tid), tm);
      return tid;
    }
    catch (RuntimeException e)
    {
      logger.fatal(Translate.get(
          "fatal.runtime.exception.requestmanager.begin", e));
      throw new SQLException(e.getMessage());
    }
  }

  public TransactionMarkerMetaData getTransactionMarker(Long tid)
      throws SQLException
  {
    TransactionMarkerMetaData tm = (TransactionMarkerMetaData) tidLoginTable
        .get(tid);

    if (tm == null)
      throw new SQLException(Translate.get("transaction.marker.not.found", ""
          + tid));

    tm.setTimeout(commitTimeout);
    return tm;
  }

  public void completeTransaction(Long tid)
  {
    tidLoginTable.remove(tid);
  }

  public void commit(long transactionId) throws SQLException
  {
    try
    {
      Long tid = new Long(transactionId);
      TransactionMarkerMetaData tm = getTransactionMarker(tid);

      // Wait for the scheduler to give us the authorization to execute
      scheduler.commit(tm);

      if (logger.isDebugEnabled())
        logger.debug(Translate.get("transaction.commit", String.valueOf(tid)));

      try
      {
        // Send to load balancer
        loadBalancer.commit(tm);

        // Notify the cache
        if (resultCache != null)
          resultCache.commit(tm.getTransactionId());

        // Notify the recovery log manager
        if (recoveryLog != null)
          recoveryLog.commit(tm);
      }
      catch (SQLException e)
      {
        throw e;
      }
      catch (AllBackendsFailedException e)
      {
        String msg = "All backends failed to commit transaction "
            + transactionId + " (" + e + ")";
        logger.error(msg);
        throw new SQLException(msg);
      }
      finally
      {
        // Notify scheduler for completion
        scheduler.commitCompleted(transactionId);

        completeTransaction(tid);
      }
    }
    catch (RuntimeException e)
    {
      logger.fatal(Translate.get(
          "fatal.runtime.exception.requestmanager.commit", e));
      throw new SQLException(e.getMessage());
    }
  }

  public void rollback(long transactionId) throws SQLException
  {
    try
    {
      Long tid = new Long(transactionId);
      TransactionMarkerMetaData tm = getTransactionMarker(tid);

      // Wait for the scheduler to give us the authorization to execute
      scheduler.rollback(tm);

      if (logger.isDebugEnabled())
        logger.debug(Translate.get("transaction.rollback", String
            .valueOf(transactionId)));

      try
      {
        // Send to load balancer
        loadBalancer.rollback(tm);

        // Send to cache
        if (this.resultCache != null)
          resultCache.rollback(transactionId);

        // Notify the recovery log manager
        if (recoveryLog != null)
        {
          recoveryLog.rollback(tm);
        }
      }
      catch (SQLException e)
      {
        throw e;
      }
      catch (AllBackendsFailedException e)
      {
        String msg = Translate.get("requestmanager.rollback.failed.all",
            new String[]{String.valueOf(transactionId), e.getMessage()});
        logger.error(msg);
        throw new SQLException(msg);
      }
      finally
      {
        // Notify scheduler for completion
        scheduler.rollbackCompleted(transactionId);

        completeTransaction(tid);
      }
    }
    catch (RuntimeException e)
    {
      logger.fatal(Translate
          .get("fatal.runtime.exception.requestmanager.rollback"), e);
      throw new SQLException(e.getMessage());
    }
  }

  //
  // Database Backends management
  //

  public void enableBackend(DatabaseBackend db) throws SQLException
  {
    loadBalancer.enableBackend(db, true);
    logger.info(Translate.get("backend.state.enabled", db.getName()));
  }

  public void enableBackendFromCheckpoint(DatabaseBackend db,
      String checkpointName) throws SQLException
  {
    // Sanity check
    if (recoveryLog == null)
    {
      String msg = Translate.get(
          "recovery.restore.checkpoint.failed.cause.null", checkpointName);
      logger.error(msg);
      throw new SQLException(msg);
    }

    db.initializeConnections();
    recoveryLog.beginRecovery();

    // Get the checkpoint from the recovery log
    long logIdx;
    try
    {
      logIdx = recoveryLog.getCheckpointRequestId(checkpointName);
    }
    catch (SQLException e)
    {
      recoveryLog.endRecovery();
      String msg = Translate.get("recovery.cannot.get.checkpoint", e);
      logger.error(msg);
      throw new SQLException(msg);
    }

    // TODO: This is not threaded yet
    JDBCRecoverThread recoverThread = new JDBCRecoverThread(recoveryLog, db,
        logIdx, loadBalancer);
    recoverThread.run();

    if (recoverThread.getException() != null)
      throw recoverThread.getException();

    // Suspend the writes
    scheduler.suspendWrites();

    // Play the remaining writes that were pending
    recoverThread.setLogIdx(recoverThread.getLogIdx() + 1);
    recoverThread.run();

    if (recoverThread.getException() != null)
    {
      scheduler.resumeWrites();
      throw recoverThread.getException();
    }

    // Now enable it
    loadBalancer.enableBackend(db, true);
    scheduler.resumeWrites();
    logger.info(Translate.get("backend.state.enabled", db.getName()));
  }

  public void disableBackend(DatabaseBackend db) throws SQLException
  {
    if (db.isReadEnabled() || db.isWriteEnabled())
    {
      loadBalancer.disableBackend(db);
      logger.info(Translate.get("backend.state.disabled", db.getName()));
    }
    else
    {
      throw new SQLException(Translate.get("backend.already.disabled", db
          .getName()));
    }
  }

  public void disableBackendForCheckpoint(DatabaseBackend db,
      String checkpointName) throws SQLException
  {
    // Sanity checks
    if (recoveryLog == null)
    {
      String msg = Translate.get("recovery.store.checkpoint.failed.cause.null",
          checkpointName);
      logger.error(msg);
      throw new SQLException(msg);
    }

    // Wait for all pending writes to finish
    logger.info(Translate.get("requestmanager.wait.pending.writes"));
    scheduler.suspendWrites();

    // Store checkpoint
    recoveryLog.storeCheckpoint(checkpointName);
    logger.info(Translate.get("recovery.checkpoint.stored", checkpointName));

    // Signal the backend should not begin any new transaction
    db.setState(BackendState.DISABLING);
    logger.info(Translate.get("backend.state.disabling", db.getName()));

    // Resume writes
    logger.info(Translate.get("requestmanager.resume.pending.writes"));
    scheduler.resumeWrites();

    // Wait for all current transactions on the backend to finish
    db.waitForAllTransactionsToComplete();

    // Now we can safely disable the backend
    db.setLastKnownCheckpoint(checkpointName);
    loadBalancer.disableBackend(db);
    logger.info(Translate.get("backend.state.disabled", db.getName()));
  }

  public void disableBackendsForCheckpoint(ArrayList backendsArrayList,
      String checkpointName) throws SQLException
  {
    // Sanity checks
    if (recoveryLog == null)
    {
      String msg = Translate.get("recovery.store.checkpoint.failed.cause.null",
          checkpointName);
      logger.error(msg);
      throw new SQLException(msg);
    }

    // Wait for all pending writes to finish
    logger.info(Translate.get("requestmanager.wait.pending.writes"));
    scheduler.suspendWrites();

    // Store checkpoint
    recoveryLog.storeCheckpoint(checkpointName);
    logger.info(Translate.get("recovery.checkpoint.stored", checkpointName));

    // Copy the list and consider only the backends that are enabled
    DatabaseBackend db;
    ArrayList backendList = (ArrayList) backendsArrayList.clone();
    for (int i = 0; i < backendList.size(); i++)
    {
      db = (DatabaseBackend) backendList.get(i);
      if (!db.isWriteEnabled())
        backendList.remove(i);
    }

    // Signal all backends that they should not begin any new transaction
    int size = backendList.size();
    for (int i = 0; i < size; i++)
    {
      db = (DatabaseBackend) backendList.get(i);
      db.setState(BackendState.DISABLING);
      logger.info(Translate.get("backend.state.disabling", db.getName()));
    }

    // Resume writes
    logger.info(Translate.get("requestmanager.resume.pending.writes"));
    scheduler.resumeWrites();

    // Wait for all current transactions on backends to finish
    for (int i = 0; i < size; i++)
    {
      db = (DatabaseBackend) backendList.get(i);
      db.waitForAllTransactionsToComplete();
    }

    // Now we can safely disable all backends
    for (int i = 0; i < size; i++)
    {
      db = (DatabaseBackend) backendList.get(i);
      db.setLastKnownCheckpoint(checkpointName);
      loadBalancer.disableBackend(db);
      logger.info(Translate.get("backend.state.disabled", db.getName()));
    }
  }

  public void callBackupManager(boolean backup, DatabaseBackend db,
      String checkpoint, ArrayList tables, boolean enableAfter,
      BackupListener listener) throws Exception
  {
    if (backup)
      backupBackendWithCheckpoint(db, checkpoint, tables, enableAfter, false,
          listener);
    else
      restoreBackendFromBackupCheckpoint(db, checkpoint, false, listener);
  }

  public void backupBackendWithCheckpoint(DatabaseBackend db,
      String checkpointName, ArrayList tables, boolean enableAfter,
      boolean wait4Result, BackupListener listener) throws SQLException
  {

    // Disable backend if needed
    if (db.isReadEnabled())
    {
      disableBackendForCheckpoint(db, checkpointName);
      logger.info(Translate.get("backend.state.disabled", db.getName()));
    }

    try
    {
      logger.info(Translate
          .get("controller.backup.octopus.start", db.getName()));
      backupManager.backup(db, checkpointName, tables, listener);
      db.setLastKnownCheckpoint(checkpointName);
      if (wait4Result)
        backupManager.getResult(db, 0);
    }
    catch (OctopusException e)
    {
      logger.error(Translate.get("controller.backup.octopus.failed"), e);
      throw new SQLException(e.getMessage());
    }
    catch (BackupException be)
    {
      logger.error(Translate.get("controller.backup.failed"), be);
      throw new SQLException(be.getMessage());
    }
    logger.info(Translate.get("controller.backup.complete", db.getName()));

    if (enableAfter)
    {
      enableBackendFromCheckpoint(db, checkpointName);
    }

  }

  public void restoreBackendFromBackupCheckpoint(DatabaseBackend db,
      String checkpointName, boolean wait4Result, BackupListener listener)
      throws OctopusException, BackupException
  {
    try
    {
      // no check for disable as we are going to overwrite
      // all the database data
      if (db.isReadEnabled())
        loadBalancer.disableBackend(db);

      backupManager.restore(db, checkpointName, null, listener);
      if (wait4Result)
        backupManager.getResult(db, 0);
    }
    catch (SQLException e1)
    {
      // This comes from the loadbalancer
      throw new BackupException(ExceptionTypes.BACKEND_CANNOT_BE_DISABLED);
    }
    catch (OctopusException e)
    {
      logger.error(Translate.get("controller.octopus.recovery.failed"), e);
      throw e;
    }
    catch (BackupException be)
    {
      logger.error(Translate.get("controller.backup.recovery.failed"), be);
      throw be;
    }
    finally
    {
      logger.info(Translate
          .get("controller.backup.recovery.done", db.getName()));
    }
  }

  public void storeBackendsInfo(String databaseName, ArrayList backends)
  {
    if (recoveryLog == null)
      return;
    int size = backends.size();
    DatabaseBackend backend;
    for (int i = 0; i < size; i++)
    {
      backend = (DatabaseBackend) backends.get(i);
      try
      {
        recoveryLog.storeBackendInfo(databaseName, backend.getName(), backend
            .getLastKnownCheckpoint(), backend.getStateValue());
      }
      catch (SQLException e)
      {
        logger.error(Translate.get("recovery.store.checkpoint.failed",
            new String[]{backend.getName(), e.getMessage()}), e);
      }
    }
  }

  public void removeCheckpoint(String checkpointName) throws SQLException
  {
    recoveryLog.removeCheckpoint(checkpointName);
  }

  //
  // Getter/Setter methods
  //

  public VirtualDatabase getVirtualDatabase()
  {
    return vdb;
  }

  public void setDatabaseSchema(DatabaseSchema schema, boolean isStatic)
  {
    if (schemaIsStatic)
    {
      if (isStatic)
      {
        logger.warn(Translate
            .get("requestmanager.schema.replace.static.with.new"));
        this.dbs = schema;
      }
      else
        logger.info(Translate.get("requestmanager.schema.ignore.new.dynamic"));
    }
    else
    {
      schemaIsStatic = isStatic;
      this.dbs = schema;
      logger.info(Translate
          .get("requestmanager.schema.set.new.virtualdatabase"));
    }

    if (schedulerParsingranularity != ParsingGranularities.NO_PARSING)
      scheduler.setDatabaseSchema(dbs);

    if (cacheParsingranularity != ParsingGranularities.NO_PARSING)
      resultCache.setDatabaseSchema(dbs);

    // Load balancers do not have a specific database schema to update
  }

  public void mergeDatabaseSchema(DatabaseSchema backendSchema)
  {
    try
    {
      if (dbs == null)
        setDatabaseSchema(new DatabaseSchema(backendSchema), false);
      else
      {
        dbs.mergeSchema(backendSchema);
        logger.info(Translate
            .get("requestmanager.schema.virtualdatabase.merged.new"));

        if (schedulerParsingranularity != ParsingGranularities.NO_PARSING)
          scheduler.mergeDatabaseSchema(dbs);

        if (cacheParsingranularity != ParsingGranularities.NO_PARSING)
          resultCache.mergeDatabaseSchema(dbs);
      }
    }
    catch (SQLException e)
    {
      logger.error(Translate.get("requestmanager.schema.merge.failed", e
          .getMessage()), e);
    }
  }

  public void setBackupManager(BackupManager currentBackupManager)
  {
    this.backupManager = currentBackupManager;
  }

  public BackupManager getBackupManager()
  {
    return backupManager;
  }

  public DatabaseSchema getDatabaseSchema()
  {
    return dbs;
  }

  public AbstractLoadBalancer getLoadBalancer()
  {
    return loadBalancer;
  }

  public void setLoadBalancer(AbstractLoadBalancer loadBalancer)
  {
    this.loadBalancer = loadBalancer;
    loadBalancerParsingranularity = loadBalancer.getParsingGranularity();
    if (loadBalancerParsingranularity > requiredGranularity)
      requiredGranularity = loadBalancerParsingranularity;
  }

  public AbstractResultCache getResultCache()
  {
    return resultCache;
  }

  public MetadataCache getMetadataCache()
  {
    return metadataCache;
  }

  public void setMetadataCache(MetadataCache metadataCache)
  {
    this.metadataCache = metadataCache;
  }

  public void setParsingCache(ParsingCache parsingCache)
  {
    parsingCache.setRequestManager(this);
    parsingCache.setGranularity(requiredGranularity);
    parsingCache.setCaseSensitiveParsing(isCaseSensitiveParsing);
    this.parsingCache = parsingCache;
  }

  public AbstractRecoveryLog getRecoveryLog()
  {
    return recoveryLog;
  }

  public void setRecoveryLog(AbstractRecoveryLog recoveryLog)
  {
    if(recoveryLog==null)
      return;
    this.recoveryLog = recoveryLog;
    ArrayList backends = vdb.getBackends();
    int size = backends.size();
    backendStateListener = new BackendStateListener(vdb.getName(), recoveryLog);
    for (int i = 0; i < size; i++)
      ((DatabaseBackend) backends.get(i))
          .setStateListener(backendStateListener);
  }

  public void setResultCache(AbstractResultCache cache)
  {
    resultCache = cache;
    cacheParsingranularity = cache.getParsingGranularity();
    if (cacheParsingranularity > requiredGranularity)
      requiredGranularity = cacheParsingranularity;
  }

  public AbstractScheduler getScheduler()
  {
    return scheduler;
  }

  public void setScheduler(AbstractScheduler scheduler)
  {
    this.scheduler = scheduler;
    schedulerParsingranularity = scheduler.getParsingGranularity();
    if (schedulerParsingranularity > requiredGranularity)
      requiredGranularity = schedulerParsingranularity;
  }

  public void setCaseSensitiveParsing(boolean isCaseSensitiveParsing)
  {
    this.isCaseSensitiveParsing = isCaseSensitiveParsing;
    if (parsingCache != null)
      parsingCache.setCaseSensitiveParsing(isCaseSensitiveParsing);
  }

  public int getRequiredParsingGranularity()
  {
    return requiredGranularity;
  }

  //
  // Debug/Monitoring
  //

  public String getXml()
  {
    StringBuffer info = new StringBuffer();
    info.append("<" + DatabasesXmlTags.ELT_RequestManager + " "
        + DatabasesXmlTags.ATT_caseSensitiveParsing + "=\""
        + isCaseSensitiveParsing + "\" " + DatabasesXmlTags.ATT_beginTimeout
        + "=\"" + beginTimeout / 1000 + "\" "
        + DatabasesXmlTags.ATT_commitTimeout + "=\"" + commitTimeout / 1000
        + "\" " + DatabasesXmlTags.ATT_rollbackTimeout + "=\""
        + rollbackTimeout / 1000 + "\">");
    if (scheduler != null)
      info.append(scheduler.getXml());

    if (metadataCache != null || parsingCache != null || resultCache != null)
    {
      info.append("<" + DatabasesXmlTags.ELT_RequestCache + ">");
      if (metadataCache != null)
        info.append(metadataCache.getXml());
      if (parsingCache != null)
        info.append(parsingCache.getXml());
      if (resultCache != null)
        info.append(resultCache.getXml());
      info.append("</" + DatabasesXmlTags.ELT_RequestCache + ">");
    }

    if (loadBalancer != null)
      info.append(loadBalancer.getXml());
    if (recoveryLog != null)
      info.append(this.recoveryLog.getXml());
    info.append("</" + DatabasesXmlTags.ELT_RequestManager + ">");
    return info.toString();
  }

  public BackendStateListener getBackendStateListener()
  {
    return backendStateListener;
  }
}

---