src/org/objectweb/cjdbc/controller/scheduler/AbstractScheduler.java

説明を見る。

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;

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
  //

  public AbstractScheduler(int raidbLevel, int parsingGranularity)
  {
    this.raidbLevel = raidbLevel;
    this.parsingGranularity = parsingGranularity;
    this.tid = 0;
    this.pendingTransactions = 0;
    this.pendingWrites = 0;
  }

  //
  // Getter/Setter methods
  //

  public final void initializeTransactionId(long transactionId)
  {
    this.tid = transactionId;
  }

  public final int getParsingGranularity()
  {
    return parsingGranularity;
  }

  public final void setParsingGranularity(int parsingGranularity)
  {
    this.parsingGranularity = parsingGranularity;
  }

  public final int getPendingWrites()
  {
    return pendingWrites;
  }

  public final int getRAIDbLevel()
  {
    return raidbLevel;
  }

  public final void setRAIDbLevel(int raidbLevel)
  {
    this.raidbLevel = raidbLevel;
  }

  public void setDatabaseSchema(DatabaseSchema dbs)
  {
    if (logger.isInfoEnabled())
      logger.info(Translate.get("scheduler.doesnt.support.schemas"));
  }

  public void mergeDatabaseSchema(DatabaseSchema dbs)
  {
    logger.info(Translate.get("scheduler.doesnt.support.schemas"));
  }

  //
  // Request Scheduling
  //

  public abstract void scheduleReadRequest(SelectRequest request)
      throws SQLException;

  public abstract void readCompletedNotify(SelectRequest request);

  public final void readCompleted(SelectRequest request)
  {
    numberRead++;
    this.readCompletedNotify(request);
  }

  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);
  }

  public abstract void scheduleNonSuspendedWriteRequest(
      AbstractWriteRequest request) throws SQLException, RollbackException;

  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++;
  }

  public abstract void notifyWriteCompleted(AbstractWriteRequest request);

  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
  //

  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;
    }
  }

  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();
        }
      }
    }
  }

  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());
  }

  protected abstract void commitTransaction(long transactionId);

  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();
        }
      }
    }
  }

  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());
  }

  protected abstract void rollbackTransaction(long transactionId);

  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
  //

  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);
      }
    }
  }

  public final void resumeNewTransactions()
  {
    synchronized (transactionSync)
    {
      suspendedTransactions = false;
      // Wake up all pending begin statements
      transactionSync.notifyAll();
    }
  }

  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);
      }
    }
  }

  public void resumeWrites()
  {
    synchronized (writesSync)
    {
      suspendedWrites = false;
      // Wake up all waiting writes
      writesSync.notifyAll();
    }
  }

  //
  // Debug/Monitoring
  //

  protected abstract String getXmlImpl();

  public String getXml()
  {
    StringBuffer info = new StringBuffer();
    info.append("<" + DatabasesXmlTags.ELT_RequestScheduler + ">");
    info.append(this.getXmlImpl());
    info.append("</" + DatabasesXmlTags.ELT_RequestScheduler + ">");
    return info.toString();
  }

  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;
  }

  public int getNumberRead()
  {
    return numberRead;
  }

  public int getNumberWrite()
  {
    return numberWrite;
  }

  public int getPendingTransactions()
  {
    return pendingTransactions;
  }
}

---