AbstractScheduler.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): _________________________.
 */

package org.objectweb.cjdbc.controller.scheduler;

import java.sql.SQLException;

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.sql.AbstractRequest;
import org.objectweb.cjdbc.common.sql.AbstractWriteRequest;
import org.objectweb.cjdbc.common.sql.SelectRequest;
import org.objectweb.cjdbc.common.sql.schema.DatabaseSchema;
import org.objectweb.cjdbc.common.xml.DatabasesXmlTags;
import org.objectweb.cjdbc.common.xml.XmlComponent;
import org.objectweb.cjdbc.controller.requestmanager.TransactionMarkerMetaData;

/**
 * The Request Scheduler should schedule the request according to a given
 * policy.
 * <p>
 * The requests comes from the Request Controller and are sent later to the next
 * ccontroller omponents (cache and load balancer).
 * 
 * @author <a href="mailto:Emmanuel.Cecchet@inria.fr">Emmanuel Cecchet </a>
 * @version 1.0
 */
public abstract class AbstractScheduler implements XmlComponent
{

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

  protected int          raidbLevel;
  protected int          parsingGranularity;

  // Transaction management
  private long           tid;
  private boolean        suspendedTransactions    = false;
  private int            pendingTransactions;
  private Object         transactionSync          = new Object();
  private Object         endOfCurrentTransactions = new Object();

  // Writes management
  private boolean        suspendedWrites          = false;
  private int            pendingWrites;
  private Object         writesSync               = new Object();
  private Object         endOfCurrentWrites       = new Object();

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

  // Monitoring values
  private int            numberRead               = 0;
  private int            numberWrite              = 0;

  //
  // Constructor
  //

  /**
   * Default scheduler to assign scheduler RAIDb level, needed granularity and
   * SQL macro handling (on the fly instanciation of NOW(), RAND(), ...).
   * 
   * @param raidbLevel RAIDb level of this scheduler
   * @param parsingGranularity Parsing granularity needed by the scheduler
   */
  public AbstractScheduler(int raidbLevel, int parsingGranularity)
  {
    this.raidbLevel = raidbLevel;
    this.parsingGranularity = parsingGranularity;
    this.tid = 0;
    this.pendingTransactions = 0;
    this.pendingWrites = 0;
  }

  //
  // Getter/Setter methods
  //

  /**
   * Initialize the transaction id with the given value (usually retrieved from
   * the recovery log).
   * 
   * @param transactionId new current transaction identifier
   */
  public final void initializeTransactionId(long transactionId)
  {
    this.tid = transactionId;
  }

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

  /**
   * Set the needed query parsing granularity.
   * 
   * @param parsingGranularity Parsing granularity needed by the scheduler
   */
  public final void setParsingGranularity(int parsingGranularity)
  {
    this.parsingGranularity = parsingGranularity;
  }

  /**
   * Returns the number of pending writes.
   * 
   * @return int
   */
  public final int getPendingWrites()
  {
    return pendingWrites;
  }

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

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

  /**
   * Sets the <code>DatabaseSchema</code> of the current virtual database.
   * This is only needed by some schedulers that will have to define their own
   * scheduler schema
   * 
   * @param dbs a <code>DatabaseSchema</code> value
   * @see org.objectweb.cjdbc.controller.scheduler.schema.SchedulerDatabaseSchema
   */
  public void setDatabaseSchema(DatabaseSchema dbs)
  {
    if (logger.isInfoEnabled())
      logger.info(Translate.get("scheduler.doesnt.support.schemas"));
  }

  /**
   * Merge the given <code>DatabaseSchema</code> with the current one.
   * 
   * @param dbs a <code>DatabaseSchema</code> value
   * @see org.objectweb.cjdbc.controller.scheduler.schema.SchedulerDatabaseSchema
   */
  public void mergeDatabaseSchema(DatabaseSchema dbs)
  {
    logger.info(Translate.get("scheduler.doesnt.support.schemas"));
  }

  //
  // Request Scheduling
  //

  /**
   * Schedule a read request (implementation specific). This method blocks until
   * the read can be executed.
   * 
   * @param request Select request to schedule (SQL macros are already handled
   *          if needed)
   * @exception SQLException if a timeout occurs
   */
  public abstract void scheduleReadRequest(SelectRequest request)
      throws SQLException;

  /**
   * Notify the completion of a read statement.
   * 
   * @param request the completed request
   */
  public abstract void readCompletedNotify(SelectRequest request);

  /**
   * Notify the completion of a read statement.
   * 
   * @param request the completed request
   */
  public final void readCompleted(SelectRequest request)
  {
    numberRead++;
    this.readCompletedNotify(request);
  }

  /**
   * Schedule a write request. This method blocks if the writes are suspended.
   * Then the number of pending writes is updated and the implementation
   * specific scheduleNonSuspendedWriteRequest function is called. SQL macros
   * are replaced in the request if the scheduler has needSQLMacroHandling set
   * to true.
   * 
   * @param request Write request to schedule
   * @exception SQLException if a timeout occurs
   * @exception RollbackException if an error occurs
   * @see #scheduleNonSuspendedWriteRequest(AbstractWriteRequest)
   */
  public final void scheduleWriteRequest(AbstractWriteRequest request)
      throws SQLException, RollbackException
  {
    synchronized (writesSync)
    {
      if (suspendedWrites)
      {
        try
        {
          // Wait on writesSync
          int timeout = request.getTimeout();
          if (timeout > 0)
          {
            long start = System.currentTimeMillis();
            long lTimeout = timeout * 1000;
            writesSync.wait(lTimeout);
            long end = System.currentTimeMillis();
            int remaining = (int) (lTimeout - (end - start));
            if (remaining > 0)
              request.setTimeout(remaining);
            else
            {
              String msg = Translate.get("scheduler.request.timeout",
                  new String[]{String.valueOf(request.getId()),
                      String.valueOf(request.getTimeout())});
              logger.warn(msg);
              throw new SQLException(msg);
            }
          }
          else
            this.writesSync.wait();
        }
        catch (InterruptedException e)
        {
          String msg = Translate.get("scheduler.request.timeout.failed", e);
          logger.warn(msg);
          throw new SQLException(msg);
        }
      }
      pendingWrites++;
    }
    scheduleNonSuspendedWriteRequest(request);
  }

  /**
   * Schedule a write request (implementation specific). This method blocks
   * until the request can be executed.
   * 
   * @param request Write request to schedule (SQL macros are already handled if
   *          needed)
   * @exception SQLException if a timeout occurs
   * @exception RollbackException if the transaction must be rollbacked
   */
  public abstract void scheduleNonSuspendedWriteRequest(
      AbstractWriteRequest request) throws SQLException, RollbackException;

  /**
   * Notify the completion of a write statement.
   * <p>
   * This method updates the number of pending writes and calls the
   * implementation specific notifyWriteCompleted function.
   * <p>
   * Finally, the suspendWrites() function is notified if needed.
   * 
   * @param request the completed request
   * @see #notifyWriteCompleted(AbstractWriteRequest)
   * @see #suspendWrites()
   */
  public final void writeCompleted(AbstractWriteRequest request)
  {
    synchronized (writesSync)
    {
      pendingWrites--;

      notifyWriteCompleted(request);

      // It this is the last write to complete and writes are
      // suspended we have to notify suspendedWrites()
      if (suspendedWrites && (pendingWrites == 0))
      {
        synchronized (endOfCurrentWrites)
        {
          endOfCurrentWrites.notifyAll();
        }
      }
    }
    numberWrite++;
  }

  /**
   * Notify the completion of a write statement. This method does not need to be
   * synchronized, it is enforced by the caller.
   * 
   * @param request the completed request
   * @see #writeCompleted(AbstractWriteRequest)
   */
  public abstract void notifyWriteCompleted(AbstractWriteRequest request);

  /**
   * Returns true if the query contains SQL macros. Currently supported macros
   * are NOW() and RAND().
   * 
   * @param request the request to check
   * @return true if the query contains SQL macros.
   */
  protected boolean hasSQLMacros(AbstractRequest request)
  {
    String lower = request.getSQL().toLowerCase();
    if (lower.indexOf("now()") > 0)
      return true;
    if (lower.indexOf("rand()") > 0)
      return true;
    return false;
  }

  //
  // Transaction management
  //

  /**
   * Begin a new transaction and return the corresponding transaction
   * identifier. This method is called from the driver when setAutoCommit(false)
   * is called.
   * 
   * @param tm The transaction marker metadata
   * @return the transaction identifier
   * @throws SQLException if an error occurs
   */
  public final long begin(TransactionMarkerMetaData tm) throws SQLException
  {
    // Check if writes are suspended
    synchronized (writesSync)
    {
      if (suspendedWrites)
      {
        try
        {
          // Wait on writesSync
          long timeout = tm.getTimeout();
          if (timeout > 0)
          {
            long start = System.currentTimeMillis();
            writesSync.wait(timeout);
            long end = System.currentTimeMillis();
            long remaining = timeout - (end - start);
            if (remaining > 0)
              tm.setTimeout(remaining);
            else
            {
              String msg = Translate.get("scheduler.begin.timeout.writeSync");
              logger.warn(msg);
              throw new SQLException(msg);
            }
          }
          else
            writesSync.wait();
        }
        catch (InterruptedException e)
        {
          String msg = Translate.get("scheduler.begin.timeout.writeSync")
              + " (" + e + ")";
          logger.error(msg);
          throw new SQLException(msg);
        }
      }
      pendingWrites++;
    }

    // Check if transactions are suspended
    synchronized (transactionSync)
    {
      if (suspendedTransactions)
        try
        {
          // Wait on transactionSync
          long timeout = tm.getTimeout();
          if (timeout > 0)
          {
            long start = System.currentTimeMillis();
            transactionSync.wait(timeout);
            long end = System.currentTimeMillis();
            long remaining = timeout - (end - start);
            if (remaining > 0)
              tm.setTimeout(remaining);
            else
            {
              String msg = Translate
                  .get("scheduler.begin.timeout.transactionSync");
              logger.warn(msg);
              throw new SQLException(msg);
            }
          }
          else
            transactionSync.wait();
        }
        catch (InterruptedException e)
        {
          String msg = Translate.get("scheduler.begin.timeout.transactionSync")
              + " (" + e + ")";
          logger.error(msg);
          throw new SQLException(msg);
        }
      tid++;
      pendingTransactions++;
      return tid;
    }
  }

  /**
   * Notify the completion of a begin command.
   * 
   * @param transactionId of the completed begin
   */
  public final void beginCompleted(long transactionId)
  {
    // Take care of suspended write
    synchronized (writesSync)
    {
      pendingWrites--;

      // It this is the last write to complete and writes are
      // suspended we have to notify suspendedWrites()
      if (suspendedWrites && (pendingWrites == 0))
      {
        synchronized (endOfCurrentWrites)
        {
          endOfCurrentWrites.notifyAll();
        }
      }
    }
  }

  /**
   * Commit a transaction.
   * <p>
   * Calls the implementation specific commitTransaction()
   * 
   * @param tm The transaction marker metadata
   * @throws SQLException if an error occurs
   * @see #commitTransaction(long)
   */
  public final void commit(TransactionMarkerMetaData tm) throws SQLException
  {
    // Check if writes are suspended
    synchronized (writesSync)
    {
      if (suspendedWrites)
      {
        try
        {
          // Wait on writesSync
          long timeout = tm.getTimeout();
          if (timeout > 0)
          {
            long start = System.currentTimeMillis();
            writesSync.wait(timeout);
            long end = System.currentTimeMillis();
            long remaining = timeout - (end - start);
            if (remaining > 0)
              tm.setTimeout(remaining);
            else
            {
              String msg = Translate.get("scheduler.commit.timeout.writeSync");
              logger.warn(msg);
              throw new SQLException(msg);
            }
          }
          else
            writesSync.wait();
        }
        catch (InterruptedException e)
        {
          String msg = Translate.get("scheduler.commit.timeout.writeSync")
              + " (" + e + ")";
          logger.error(msg);
          throw new SQLException(msg);
        }
      }
      pendingWrites++;
    }
    commitTransaction(tm.getTransactionId());
  }

  /**
   * Commit a transaction given its id.
   * 
   * @param transactionId the transaction id
   * @exception RemoteException if an error occurs
   */
  protected abstract void commitTransaction(long transactionId);

  /**
   * Notify the completion of a commit command.
   * 
   * @param transactionId of the completed commit
   */
  public final void commitCompleted(long transactionId)
  {
    // Take care of suspended transactions
    synchronized (transactionSync)
    {
      pendingTransactions--;

      // If it is the last pending transaction to complete and we
      // are waiting for pending transactions to complete, then wake
      // up suspendNewTransactionsForCheckpoint()
      if (suspendedTransactions && (pendingTransactions == 0))
      {
        synchronized (endOfCurrentTransactions)
        {
          endOfCurrentTransactions.notifyAll();
        }
      }
    }
    // Take care of suspended write
    synchronized (writesSync)
    {
      pendingWrites--;

      // It this is the last write to complete and writes are
      // suspended we have to notify suspendedWrites()
      if (suspendedWrites && (pendingWrites == 0))
      {
        synchronized (endOfCurrentWrites)
        {
          endOfCurrentWrites.notifyAll();
        }
      }
    }
  }

  /**
   * Rollback a transaction.
   * <p>
   * Calls the implementation specific rollbackTransaction()
   * 
   * @param tm The transaction marker metadata
   * @exception SQLException if an error occurs
   * @see #rollbackTransaction(long)
   */
  public final void rollback(TransactionMarkerMetaData tm) throws SQLException
  {
    // Check if writes are suspended
    synchronized (writesSync)
    {
      if (suspendedWrites)
      {
        try
        {
          // Wait on writesSync
          long timeout = tm.getTimeout();
          if (timeout > 0)
          {
            long start = System.currentTimeMillis();
            writesSync.wait(timeout);
            long end = System.currentTimeMillis();
            long remaining = timeout - (end - start);
            if (remaining > 0)
              tm.setTimeout(remaining);
            else
            {
              String msg = Translate
                  .get("scheduler.rollback.timeout.writeSync");
              logger.warn(msg);
              throw new SQLException(msg);
            }
          }
          else
            writesSync.wait();
        }
        catch (InterruptedException e)
        {
          String msg = Translate.get("scheduler.rollback.timeout.writeSync")
              + " (" + e + ")";
          logger.error(msg);
          throw new SQLException(msg);
        }
      }
      pendingWrites++;
    }
    rollbackTransaction(tm.getTransactionId());
  }

  /**
   * Rollback a transaction given its id.
   * 
   * @param transactionId the transaction id
   * @exception RemoteException if an error occurs
   */
  protected abstract void rollbackTransaction(long transactionId);

  /**
   * Notify the completion of a rollback command.
   * 
   * @param transactionId of the rollback commit
   */
  public final void rollbackCompleted(long transactionId)
  {
    // Take care of suspended transactions
    synchronized (transactionSync)
    {
      pendingTransactions--;

      // If it is the last pending transaction to complete and we
      // are waiting for pending transactions to complete, then wake
      // up suspendNewTransactionsForCheckpoint()
      if (suspendedTransactions && (pendingTransactions == 0))
      {
        synchronized (endOfCurrentTransactions)
        {
          endOfCurrentTransactions.notifyAll();
        }
      }
    }
    // Take care of suspended write
    synchronized (writesSync)
    {
      pendingWrites--;

      // It this is the last write to complete and writes are
      // suspended we have to notify suspendedWrites()
      if (suspendedWrites && (pendingWrites == 0))
      {
        synchronized (endOfCurrentWrites)
        {
          endOfCurrentWrites.notifyAll();
        }
      }
    }
  }

  //
  // Checkpoint management
  //

  /**
   * Suspend all calls to begin() until all current transactions are finished in
   * order to store a checkpoint. This method returns when all pending
   * transactions have finished.
   * <p>
   * New transactions remain suspended until resumeNewTransactions() is called.
   * 
   * @throws SQLException if an error occurs
   * @see #resumeNewTransactions()
   */
  public final void suspendNewTransactionsForCheckpoint() throws SQLException
  {
    synchronized (transactionSync)
    {
      suspendedTransactions = true;
      if (pendingTransactions == 0)
        return;
    }

    synchronized (endOfCurrentTransactions)
    {
      // Here we have a potential synchronization problem since the last
      // transaction completion could have happened before we entered this
      // synchronized block. Therefore we recheck if there is effectively
      // still pending transactions. If this is not the case, we don't have
      // to sleep and we can immediately return.
      if (pendingTransactions == 0)
        return;

      // Wait for pending transactions to end
      try
      {
        endOfCurrentTransactions.wait();
      }
      catch (InterruptedException e)
      {
        String msg = Translate.get("scheduler.suspend.transaction.failed", e);
        logger.error(msg);
        throw new SQLException(msg);
      }
    }
  }

  /**
   * Resume new transactions that were suspended by
   * suspendNewTransactionsForCheckpoint().
   * 
   * @see #suspendNewTransactionsForCheckpoint()
   */
  public final void resumeNewTransactions()
  {
    synchronized (transactionSync)
    {
      suspendedTransactions = false;
      // Wake up all pending begin statements
      transactionSync.notifyAll();
    }
  }

  /**
   * Suspend all write queries. This method blocks until all pending writes are
   * completed.
   * <p>
   * Writes execution is resumed by calling resumeWrites()
   * 
   * @throws SQLException if an error occurs
   * @see #resumeWrites()
   */
  public void suspendWrites() throws SQLException
  {
    synchronized (writesSync)
    {
      suspendedWrites = true;
      if (pendingWrites == 0)
        return;
    }

    synchronized (endOfCurrentWrites)
    {
      // Here we have a potential synchronization problem since the last
      // write completion could have happened before we entered this
      // synchronized block. Therefore we recheck if there is effectively
      // still pending writes. If this is not the case, we don't have
      // to sleep and we can immediately return.
      if (pendingWrites == 0)
        return;

      // Wait for pending transactions to end
      try
      {
        endOfCurrentWrites.wait();
      }
      catch (InterruptedException e)
      {
        String msg = Translate.get("scheduler.suspend.writes.failed", e);
        logger.error(msg);
        throw new SQLException(msg);
      }
    }
  }

  /**
   * Resume the execution of write queries that were suspended by
   * suspendWrites().
   * 
   * @see #suspendWrites()
   */
  public void resumeWrites()
  {
    synchronized (writesSync)
    {
      suspendedWrites = false;
      // Wake up all waiting writes
      writesSync.notifyAll();
    }
  }

  //
  // Debug/Monitoring
  //

  protected abstract String getXmlImpl();

  /**
   * Get information about the Request Scheduler in xml format
   * 
   * @return <code>String</code> containing information in xml
   */
  public String getXml()
  {
    StringBuffer info = new StringBuffer();
    info.append("<" + DatabasesXmlTags.ELT_RequestScheduler + ">");
    info.append(this.getXmlImpl());
    info.append("</" + DatabasesXmlTags.ELT_RequestScheduler + ">");
    return info.toString();
  }

  /**
   * Returns live information on the scheduler
   * 
   * @return array of data
   */
  public String[] getSchedulerData()
  {
    String[] data = new String[7];
    data[0] = "" + numberRead;
    data[1] = "" + numberWrite;
    data[2] = "" + pendingTransactions;
    data[3] = "" + pendingWrites;
    data[4] = "" + numberRead + numberWrite;
    data[5] = (suspendedTransactions) ? "1" : "0";
    data[6] = (suspendedWrites) ? "1" : "0";
    return data;
  }

  /**
   * @return Returns the numberRead.
   */
  public int getNumberRead()
  {
    return numberRead;
  }

  /**
   * @return Returns the numberWrite.
   */
  public int getNumberWrite()
  {
    return numberWrite;
  }

  /**
   * @return Returns the pendingTransactions.
   */
  public int getPendingTransactions()
  {
    return pendingTransactions;
  }
  /**
   * @return Returns the suspendedTransactions.
   */
  public boolean isSuspendedTransactions()
  {
    return suspendedTransactions;
  }
  /**
   * @return Returns the suspendedWrites.
   */
  public boolean isSuspendedWrites()
  {
    return suspendedWrites;
  }
}

---