src/org/objectweb/cjdbc/controller/scheduler/raidb1/RAIDb1OptimisticTransactionLevelScheduler.java

説明を見る。

package org.objectweb.cjdbc.controller.scheduler.raidb1;

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

import org.objectweb.cjdbc.common.exceptions.RollbackException;
import org.objectweb.cjdbc.common.sql.AbstractWriteRequest;
import org.objectweb.cjdbc.common.sql.ParsingGranularities;
import org.objectweb.cjdbc.common.sql.SelectRequest;
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.controller.requestmanager.RAIDbLevels;
import org.objectweb.cjdbc.controller.scheduler.AbstractScheduler;
import org.objectweb.cjdbc.controller.scheduler.schema.SchedulerDatabaseSchema;
import org.objectweb.cjdbc.controller.scheduler.schema.SchedulerDatabaseTable;
import org.objectweb.cjdbc.controller.scheduler.schema.TransactionExclusiveLock;

public class RAIDb1OptimisticTransactionLevelScheduler
    extends
      AbstractScheduler
{

  //
  // How the code is organized ?
  //
  // 1. Member variables
  // 2. Constructor
  // 3. Request handling
  // 4. Transaction management
  // 5. Debug/Monitoring
  //

  private long                    requestId;
  private SchedulerDatabaseSchema schedulerDatabaseSchema = null;

  //
  // Constructor
  //

  public RAIDb1OptimisticTransactionLevelScheduler()
  {
    super(RAIDbLevels.RAIDb1, ParsingGranularities.TABLE);
    requestId = 0;
  }

  //
  // Request Handling
  //

  public synchronized void setDatabaseSchema(DatabaseSchema dbs)
  {
    if (schedulerDatabaseSchema == null)
    {
      logger.info("Setting new database schema");
      schedulerDatabaseSchema = new SchedulerDatabaseSchema(dbs);
    }
    else
    { // Schema is updated, compute the diff !
      SchedulerDatabaseSchema newSchema = new SchedulerDatabaseSchema(dbs);
      ArrayList tables = schedulerDatabaseSchema.getTables();
      ArrayList newTables = newSchema.getTables();
      if (newTables == null)
      { // New schema is empty (no backend is active anymore)
        logger.info("Removing all tables.");
        schedulerDatabaseSchema = null;
        return;
      }

      // Remove extra-tables
      for (int i = 0; i < tables.size(); i++)
      {
        SchedulerDatabaseTable t = (SchedulerDatabaseTable) tables.get(i);
        if (!newSchema.hasTable(t.getName()))
        {
          schedulerDatabaseSchema.removeTable(t);
          if (logger.isInfoEnabled())
            logger.info("Removing table " + t.getName());
        }
      }

      // Add missing tables
      int size = newTables.size();
      for (int i = 0; i < size; i++)
      {
        SchedulerDatabaseTable t = (SchedulerDatabaseTable) newTables.get(i);
        if (!schedulerDatabaseSchema.hasTable(t.getName()))
        {
          schedulerDatabaseSchema.addTable(t);
          if (logger.isInfoEnabled())
            logger.info("Adding table " + t.getName());
        }
      }
    }
  }

  public void mergeDatabaseSchema(DatabaseSchema dbs)
  {
    try
    {
      logger.info("Merging new database schema");
      schedulerDatabaseSchema.mergeSchema(new SchedulerDatabaseSchema(dbs));
    }
    catch (Exception e)
    {
      logger.error("Error while merging new database schema", e);
    }
  }

  public final void scheduleReadRequest(SelectRequest request)
      throws SQLException
  {
    synchronized (this)
    {
      request.setId(requestId++);
    }
  }

  public final void readCompletedNotify(SelectRequest request)
  {
  }

  public void scheduleNonSuspendedWriteRequest(AbstractWriteRequest request)
      throws SQLException, RollbackException
  {
    if (request.isCreate())
    {
      synchronized (this)
      {
        request.setId(requestId++);
      }
      return;
    }

    SchedulerDatabaseTable t = schedulerDatabaseSchema.getTable(request
        .getTableName());
    if (t == null)
    {
      String msg = "No table found for request " + request.getId();
      logger.error(msg);
      throw new SQLException(msg);
    }

    // Deadlock detection
    TransactionExclusiveLock tableLock = t.getLock();
    if (!request.isAutoCommit())
    {
      synchronized (this)
      {
        if (tableLock.isLocked())
        { // Is the lock owner blocked by a lock we already own?
          long owner = tableLock.getLocker();
          long us = request.getTransactionId();
          if (owner != us)
          { // Parse all tables
            ArrayList tables = schedulerDatabaseSchema.getTables();
            ArrayList weAreblocking = new ArrayList();
            int size = tables.size();
            for (int i = 0; i < size; i++)
            {
              SchedulerDatabaseTable table = (SchedulerDatabaseTable) tables
                  .get(i);
              if (table == null)
                continue;
              TransactionExclusiveLock lock = table.getLock();
              // Are we the lock owner ?
              if (lock.isLocked())
              {
                if (lock.getLocker() == us)
                {
                  // Is 'owner' in the list of the blocked transactions?
                  if (lock.isWaiting(owner))
                  { // Deadlock detected, we must rollback
                    releaseLocks(us);
                    throw new RollbackException(
                        "Deadlock detected, rollbacking transaction " + us);
                  }
                  else
                    weAreblocking.addAll(lock.getWaitingList());
                }
              }
            }
          }
          else
          { // We are the lock owner and already synchronized on this
            // Assign the request id and exit
            request.setId(requestId++);
            return;
          }
        }
        else
        { // Lock is free, take it in the synchronized block
          acquireLockAndSetRequestId(request, tableLock);
          return;
        }
      }
    }

    acquireLockAndSetRequestId(request, tableLock);
  }

  private void acquireLockAndSetRequestId(AbstractWriteRequest request,
      TransactionExclusiveLock tableLock) throws SQLException
  {
    // Acquire the lock
    if (tableLock.acquire(request))
    {
      synchronized (this)
      {
        request.setId(requestId++);
      }
      if (logger.isDebugEnabled())
        logger.debug("Request " + request.getId() + " scheduled for write ("
            + getPendingWrites() + " pending writes)");
    }
    else
    {
      if (logger.isWarnEnabled())
        logger.warn("Request " + request.getId() + " timed out ("
            + request.getTimeout() + " ms)");
      throw new SQLException("Timeout (" + request.getTimeout()
          + ") for request: " + request.getId());
    }
  }

  public final void notifyWriteCompleted(AbstractWriteRequest request)
  {
    if (request.isCreate())
    { // Add table to schema
      if (logger.isDebugEnabled())
        logger.debug("Adding table '" + request.getTableName()
            + "' to scheduler schema");
      synchronized (this)
      {
        schedulerDatabaseSchema.addTable(new SchedulerDatabaseTable(
            new DatabaseTable(request.getTableName())));
      }
    }
    else if (request.isDrop())
    { // Drop table from schema
      if (logger.isDebugEnabled())
        logger.debug("Removing table '" + request.getTableName()
            + "' to scheduler schema");
      synchronized (this)
      {
        schedulerDatabaseSchema.removeTable(schedulerDatabaseSchema
            .getTable(request.getTableName()));
      }
      return;
    }

    // Requests outside transaction delimiters must release the lock
    // as soon as they have executed
    if (request.isAutoCommit())
    {
      SchedulerDatabaseTable t = schedulerDatabaseSchema.getTable(request
          .getTableName());
      if (t == null)
      {
        String msg = "No table found to release lock for request "
            + request.getId();
        logger.error(msg);
      }
      else
        t.getLock().release();
    }
  }

  //
  // Transaction Management
  //

  protected final void commitTransaction(long transactionId)
  {
    releaseLocks(transactionId);
  }

  protected final void rollbackTransaction(long transactionId)
  {
    releaseLocks(transactionId);
  }

  private synchronized void releaseLocks(long transactionId)
  {
    ArrayList tables = schedulerDatabaseSchema.getTables();
    int size = tables.size();
    for (int i = 0; i < size; i++)
    {
      SchedulerDatabaseTable t = (SchedulerDatabaseTable) tables.get(i);
      if (t == null)
        continue;
      TransactionExclusiveLock lock = t.getLock();
      // Are we the lock owner ?
      if (lock.isLocked())
        if (lock.getLocker() == transactionId)
          lock.release();
    }
  }

  //
  // Debug/Monitoring
  //

  public String getXmlImpl()
  {
    return "<" + DatabasesXmlTags.ELT_RAIDb1Scheduler + " "
        + DatabasesXmlTags.ATT_level + "=\""
        + DatabasesXmlTags.VAL_optimisticTransaction + "\"/>";
  }

}

---