src/org/objectweb/cjdbc/controller/requestmanager/distributed/RAIDb1DistributedRequestManager.java

説明を見る。

package org.objectweb.cjdbc.controller.requestmanager.distributed;

import java.sql.ResultSet;
import java.sql.SQLException;
import java.util.Vector;

import org.jgroups.Address;
import org.jgroups.blocks.GroupRequest;
import org.jgroups.util.RspList;
import org.objectweb.cjdbc.common.i18n.Translate;
import org.objectweb.cjdbc.common.sql.AbstractRequest;
import org.objectweb.cjdbc.common.sql.AbstractWriteRequest;
import org.objectweb.cjdbc.common.sql.StoredProcedure;
import org.objectweb.cjdbc.common.sql.UnknownRequest;
import org.objectweb.cjdbc.controller.cache.result.AbstractResultCache;
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.scheduler.AbstractScheduler;
import org.objectweb.cjdbc.controller.virtualdatabase.ControllerResultSet;
import org.objectweb.cjdbc.controller.virtualdatabase.DistributedVirtualDatabase;
import org.objectweb.cjdbc.controller.virtualdatabase.protocol.CJDBCGroupMessage;
import org.objectweb.cjdbc.controller.virtualdatabase.protocol.Commit;
import org.objectweb.cjdbc.controller.virtualdatabase.protocol.ExecReadStoredProcedure;
import org.objectweb.cjdbc.controller.virtualdatabase.protocol.ExecWriteRequest;
import org.objectweb.cjdbc.controller.virtualdatabase.protocol.ExecWriteRequestWithKeys;
import org.objectweb.cjdbc.controller.virtualdatabase.protocol.ExecWriteStoredProcedure;
import org.objectweb.cjdbc.controller.virtualdatabase.protocol.NotifyCompletion;
import org.objectweb.cjdbc.controller.virtualdatabase.protocol.Rollback;

public class RAIDb1DistributedRequestManager extends DistributedRequestManager
{
  // Used to check if a result was received or not
  private static final int NO_RESULT = -5;

  public RAIDb1DistributedRequestManager(DistributedVirtualDatabase vdb,
      AbstractScheduler scheduler, AbstractResultCache cache,
      AbstractLoadBalancer loadBalancer, AbstractRecoveryLog recoveryLog,
      long beginTimeout, long commitTimeout, long rollbackTimeout)
      throws SQLException
  {
    super(vdb, scheduler, cache, loadBalancer, recoveryLog, beginTimeout,
        commitTimeout, rollbackTimeout);
  }

  public int execDistributedWriteRequest(AbstractWriteRequest request)
      throws SQLException
  {
    try
    {
      int execWriteRequestResult = -1;

      Vector groupMembers = dvdb.getCurrentView().getMembers();

      if (logger.isDebugEnabled())
        logger.debug("Broadcasting request "
            + request.getSQLShortForm(dvdb.getSQLShortFormLength())
            + (request.isAutoCommit() ? "" : " transaction "
                + request.getTransactionId()) + " to all controllers ("
            + dvdb.getChannel().getLocalAddress() + "->"
            + groupMembers.toString() + ")");

      // Send the query to everybody including us
      RspList responses = dvdb.getDispatcher().castMessage(groupMembers,
          CJDBCGroupMessage.getMessage(new ExecWriteRequest(request)),
          GroupRequest.GET_ALL, request.getTimeout());

      if (logger.isDebugEnabled())
        logger.debug("Request "
            + request.getSQLShortForm(dvdb.getSQLShortFormLength())
            + " completed.");

      if (responses.numSuspectedMembers() > 0)
      { // Some controllers failed ... too bad !
        logger.warn(responses.numSuspectedMembers()
            + " controller(s) died during execution of request "
            + request.getId());
      }

      // List of controllers that gave a AllBackendsFailedException
      Vector failedOnAllBackends = null;
      SQLException exception = null;
      int size = groupMembers.size();
      // Get the result of each controller
      for (int i = 0; i < size; i++)
      {
        Address address = (Address) groupMembers.get(i);
        if (responses.isSuspected(address))
        {
          logger.warn("Controller " + address + " is suspected of failure.");
          continue;
        }
        Object r = responses.get(address);
        if (r instanceof Integer)
        {
          if (execWriteRequestResult == -1)
            execWriteRequestResult = ((Integer) r).intValue();
          else if (execWriteRequestResult != ((Integer) r).intValue())
            logger.error("Controllers have different results for request "
                + request.getId());
        }
        else if (r instanceof AllBackendsFailedException)
        {
          if (failedOnAllBackends == null)
            failedOnAllBackends = new Vector();
          failedOnAllBackends.add(address);
          if (logger.isDebugEnabled())
            logger.debug("Request failed on all backends of controller "
                + address + " (" + r + ")");
        }
        else if (r instanceof SQLException)
        {
          String msg = "Request " + request.getId() + " failed on controller "
              + address + " (" + r + ")";
          logger.warn(msg);
          exception = (SQLException) r;
        }
      }

      if (failedOnAllBackends != null)
      { // Notify all controllers of completion
        dvdb.getDispatcher().castMessage(
            failedOnAllBackends,
            CJDBCGroupMessage.getMessage(new NotifyCompletion(request,
                execWriteRequestResult != -1)), GroupRequest.GET_NONE,
            request.getTimeout());
      }

      if (execWriteRequestResult != -1)
      {
        if (logger.isDebugEnabled())
          logger.debug("Request " + request.getId()
              + " completed successfully.");
        return execWriteRequestResult;
      }
      else if (exception != null)
        throw exception;

      // At this point, all controllers failed
      String msg = "Request '" + request + "' failed on all controllers";
      logger.warn(msg);
      throw new SQLException(msg);
    }
    catch (SQLException e)
    {
      String msg = Translate
          .get("loadbalancer.request.failed", new String[]{
              request.getSQLShortForm(vdb.getSQLShortFormLength()),
              e.getMessage()});
      logger.warn(msg);
      throw e;
    }
  }

  public ControllerResultSet execDistributedWriteRequestWithKeys(
      AbstractWriteRequest request) throws SQLException
  {
    try
    {
      ControllerResultSet execWriteRequestResult = null;

      Vector groupMembers = dvdb.getCurrentView().getMembers();

      if (logger.isDebugEnabled())
        logger.debug("Broadcasting request "
            + request.getSQLShortForm(dvdb.getSQLShortFormLength())
            + (request.isAutoCommit() ? "" : " transaction "
                + request.getTransactionId()) + ") to all controllers ("
            + dvdb.getChannel().getLocalAddress() + "->"
            + groupMembers.toString() + ")");

      // Send the query to everybody including us
      RspList responses = dvdb.getDispatcher().castMessage(groupMembers,
          CJDBCGroupMessage.getMessage(new ExecWriteRequestWithKeys(request)),
          GroupRequest.GET_ALL, request.getTimeout());

      if (logger.isDebugEnabled())
        logger.debug("Request "
            + request.getSQLShortForm(dvdb.getSQLShortFormLength())
            + " completed.");

      if (responses.numSuspectedMembers() > 0)
      { // Some controllers failed ... too bad !
        logger.warn(responses.numSuspectedMembers()
            + " controller(s) died during execution of request "
            + request.getId());
      }

      // List of controllers that gave a AllBackendsFailedException
      Vector failedOnAllBackends = null;
      SQLException exception = null;
      int size = groupMembers.size();
      // Get the result of each controller
      for (int i = 0; i < size; i++)
      {
        Address address = (Address) groupMembers.get(i);
        if (responses.isSuspected(address))
        {
          logger.warn("Controller " + address + " is suspected of failure.");
          continue;
        }
        Object r = responses.get(address);
        if (r instanceof ResultSet)
        {
          if (execWriteRequestResult == null)
            execWriteRequestResult = (ControllerResultSet) r;
        }
        else if (r instanceof AllBackendsFailedException)
        {
          if (failedOnAllBackends == null)
            failedOnAllBackends = new Vector();
          failedOnAllBackends.add(address);
          if (logger.isDebugEnabled())
            logger.debug("Request failed on all backends of controller "
                + address + " (" + r + ")");
        }
        else if (r instanceof SQLException)
        {
          String msg = "Request " + request.getId() + " failed on controller "
              + address + " (" + r + ")";
          logger.warn(msg);
          exception = (SQLException) r;
        }
      }

      if (failedOnAllBackends != null)
      { // Notify all controllers of completion
        dvdb.getDispatcher().castMessage(
            failedOnAllBackends,
            CJDBCGroupMessage.getMessage(new NotifyCompletion(request,
                execWriteRequestResult != null)), GroupRequest.GET_NONE,
            request.getTimeout());
      }

      if (execWriteRequestResult != null)
      {
        if (logger.isDebugEnabled())
          logger.debug("Request " + request.getId()
              + " completed successfully.");
        return execWriteRequestResult;
      }
      else if (exception != null)
        throw exception;
      // At this point, all controllers failed
      String msg = "Request '" + request + "' failed on all controllers";
      logger.warn(msg);
      throw new SQLException(msg);
    }
    catch (SQLException e)
    {
      String msg = Translate
          .get("loadbalancer.request.failed", new String[]{
              request.getSQLShortForm(vdb.getSQLShortFormLength()),
              e.getMessage()});
      logger.warn(msg);
      throw e;
    }
  }

  public ControllerResultSet execDistributedReadStoredProcedure(
      StoredProcedure proc) throws SQLException
  {
    try
    {
      ControllerResultSet result = null;

      Vector groupMembers = dvdb.getCurrentView().getMembers();

      if (logger.isDebugEnabled())
        logger.debug("Broadcasting read stored procedure "
            + proc.getSQLShortForm(dvdb.getSQLShortFormLength())
            + (proc.isAutoCommit() ? "" : " transaction "
                + proc.getTransactionId()) + ") to all controllers ("
            + dvdb.getChannel().getLocalAddress() + "->"
            + groupMembers.toString() + ")");

      // Send the query to everybody including us
      RspList responses = dvdb.getDispatcher().castMessage(groupMembers,
          CJDBCGroupMessage.getMessage(new ExecReadStoredProcedure(proc)),
          GroupRequest.GET_ALL, proc.getTimeout());

      if (logger.isDebugEnabled())
        logger.debug("Stored procedure "
            + proc.getSQLShortForm(dvdb.getSQLShortFormLength())
            + " completed.");

      if (responses.numSuspectedMembers() > 0)
      { // Some controllers failed ... too bad !
        logger.warn(responses.numSuspectedMembers()
            + " controller(s) died during execution of stored procedure "
            + proc.getId());
      }

      // List of controllers that gave a AllBackendsFailedException
      Vector failedOnAllBackends = null;
      SQLException exception = null;
      int size = groupMembers.size();
      // Get the result of each controller
      for (int i = 0; i < size; i++)
      {
        Address address = (Address) groupMembers.get(i);
        if (responses.isSuspected(address))
        {
          logger.warn("Controller " + address + " is suspected of failure.");
          continue;
        }
        Object r = responses.get(address);
        if (r instanceof ControllerResultSet)
        {
          if (result == null)
            result = (ControllerResultSet) r;
        }
        else if (r instanceof AllBackendsFailedException)
        {
          if (failedOnAllBackends == null)
            failedOnAllBackends = new Vector();
          failedOnAllBackends.add(address);
        }
        else if (r instanceof SQLException)
        {
          String msg = "Stored procedure " + proc.getId()
              + " failed on controller " + address + " (" + r + ")";
          logger.warn(msg);
          exception = (SQLException) r;
        }
      }

      if (failedOnAllBackends != null)
      { // Notify all controllers of completion
        dvdb.getDispatcher().castMessage(
            failedOnAllBackends,
            CJDBCGroupMessage.getMessage(new NotifyCompletion(proc,
                result != null)), GroupRequest.GET_NONE, proc.getTimeout());
      }

      if (result != null)
      {
        if (logger.isDebugEnabled())
          logger.debug("Stored procedure " + proc.getId()
              + " completed successfully.");
        return result; // Success
      }
      else if (exception != null)
        throw exception;

      // At this point, all controllers failed
      String msg = "Stored procedure '" + proc + "' failed on all controllers";
      logger.warn(msg);
      throw new SQLException(msg);
    }
    catch (SQLException e)
    {
      String msg = Translate.get("loadbalancer.request.failed", new String[]{
          proc.getSQLShortForm(vdb.getSQLShortFormLength()), e.getMessage()});
      logger.warn(msg);
      throw e;
    }
  }

  public int execDistributedWriteStoredProcedure(StoredProcedure proc)
      throws SQLException
  {
    try
    {
      int execWriteStoredProcedureResult = NO_RESULT;

      Vector groupMembers = dvdb.getCurrentView().getMembers();

      if (logger.isDebugEnabled())
        logger.debug("Broadcasting write store procedure "
            + proc.getSQLShortForm(dvdb.getSQLShortFormLength())
            + (proc.isAutoCommit() ? "" : " transaction "
                + proc.getTransactionId()) + ") to all controllers ("
            + dvdb.getChannel().getLocalAddress() + "->"
            + groupMembers.toString() + ")");

      // Send the query to everybody including us
      RspList responses = dvdb.getDispatcher().castMessage(groupMembers,
          CJDBCGroupMessage.getMessage(new ExecWriteStoredProcedure(proc)),
          GroupRequest.GET_ALL, proc.getTimeout());

      if (logger.isDebugEnabled())
        logger.debug("Stored procedure "
            + proc.getSQLShortForm(dvdb.getSQLShortFormLength())
            + " completed.");

      if (responses.numSuspectedMembers() > 0)
      { // Some controllers failed ... too bad !
        logger.warn(responses.numSuspectedMembers()
            + " controller(s) died during execution of stored procedure "
            + proc.getId());
      }

      // List of controllers that gave a AllBackendsFailedException
      Vector failedOnAllBackends = null;
      SQLException exception = null;
      int size = groupMembers.size();
      // Get the result of each controller
      for (int i = 0; i < size; i++)
      {
        Address address = (Address) groupMembers.get(i);
        if (responses.isSuspected(address))
        {
          logger.warn("Controller " + address + " is suspected of failure.");
          continue;
        }
        Object r = responses.get(address);
        if (r instanceof Integer)
        {
          if (execWriteStoredProcedureResult == NO_RESULT)
            execWriteStoredProcedureResult = ((Integer) r).intValue();
        }
        else if (r instanceof AllBackendsFailedException)
        {
          if (failedOnAllBackends == null)
            failedOnAllBackends = new Vector();
          failedOnAllBackends.add(address);
        }
        else if (r instanceof SQLException)
        {
          String msg = "Stored procedure " + proc.getId()
              + " failed on controller " + address + " (" + r + ")";
          logger.warn(msg);
          exception = (SQLException) r;
        }
      }

      if (failedOnAllBackends != null)
      { // Notify all controllers of completion
        dvdb.getDispatcher().castMessage(
            failedOnAllBackends,
            CJDBCGroupMessage.getMessage(new NotifyCompletion(proc,
                execWriteStoredProcedureResult != NO_RESULT)),
            GroupRequest.GET_NONE, proc.getTimeout());
      }

      if (execWriteStoredProcedureResult != NO_RESULT)
      {
        if (logger.isDebugEnabled())
          logger.debug("Stored procedure " + proc.getId()
              + " completed successfully.");
        return execWriteStoredProcedureResult; // Success
      }
      else if (exception != null)
        throw exception;

      // At this point, all controllers failed
      String msg = "Stored procedure '" + proc + "' failed on all controllers";
      logger.warn(msg);
      throw new SQLException(msg);
    }
    catch (SQLException e)
    {
      String msg = Translate.get("loadbalancer.request.failed", new String[]{
          proc.getSQLShortForm(vdb.getSQLShortFormLength()), e.getMessage()});
      logger.warn(msg);
      throw e;
    }
  }

  public void distributedCommit(long transactionId) throws SQLException
  {
    try
    {
      Vector groupMembers = dvdb.getCurrentView().getMembers();
      if (logger.isDebugEnabled())
        logger.debug("Broadcasting transaction " + transactionId
            + " commit to all controllers ("
            + dvdb.getChannel().getLocalAddress() + "->"
            + groupMembers.toString() + ")");

      // Send the query to everybody including us
      RspList responses = dvdb.getDispatcher().castMessage(groupMembers,
          CJDBCGroupMessage.getMessage(new Commit(transactionId)),
          GroupRequest.GET_ALL, this.commitTimeout);

      if (logger.isDebugEnabled())
        logger.debug("Commit of transaction " + transactionId + " completed.");

      if (responses.numSuspectedMembers() > 0)
      { // Some controllers failed ... too bad !
        logger.warn(responses.numSuspectedMembers()
            + " controller(s) died during execution of commit for transaction "
            + transactionId);
      }

      // List of controllers that gave a AllBackendsFailedException
      Vector failedOnAllBackends = null;
      SQLException exception = null;
      int size = groupMembers.size();
      boolean success = false;
      // Get the result of each controller
      for (int i = 0; i < size; i++)
      {
        Address address = (Address) groupMembers.get(i);
        if (responses.isSuspected(address))
        {
          logger.warn("Controller " + address + " is suspected of failure.");
          continue;
        }
        Object r = responses.get(address);
        if (r instanceof Boolean)
        {
          if (((Boolean) r).booleanValue())
            success = true;
          else
            logger.error("Unexpected result for controller  " + address);
        }
        else if (r instanceof AllBackendsFailedException)
        {
          if (failedOnAllBackends == null)
            failedOnAllBackends = new Vector();
          failedOnAllBackends.add(address);
          if (logger.isDebugEnabled())
            logger.debug("Commit failed on all backends of controller "
                + address + " (" + r + ")");
        }
        else if (r instanceof SQLException)
        {
          String msg = "Commit of transaction " + transactionId
              + " failed on controller " + address + " (" + r + ")";
          logger.warn(msg);
          exception = (SQLException) r;
        }
      }

      if (failedOnAllBackends != null)
      { // Notify all controllers where all backend failed that the commit
        // completed with 'success'
        AbstractRequest request = new UnknownRequest("commit", false, 0, "\n");
        request.setTransactionId(transactionId);
        dvdb.getDispatcher().castMessage(
            failedOnAllBackends,
            CJDBCGroupMessage
                .getMessage(new NotifyCompletion(request, success)),
            GroupRequest.GET_NONE, commitTimeout);
      }

      if (success)
        return; // This is a success if at least one controller has succeeded

      if (exception != null)
        throw exception;
      // At this point, all controllers failed
      String msg = "Transaction " + transactionId
          + " failed to commit on all controllers";
      logger.warn(msg);
      throw new SQLException(msg);
    }
    catch (SQLException e)
    {
      String msg = "Transaction " + transactionId + " commit failed (" + e
          + ")";
      logger.warn(msg);
      throw e;
    }
  }

  public void distributedRollback(long transactionId) throws SQLException
  {
    try
    {
      Vector groupMembers = dvdb.getCurrentView().getMembers();
      if (logger.isDebugEnabled())
        logger.debug("Broadcasting transaction " + transactionId
            + " rollback to all controllers ("
            + dvdb.getChannel().getLocalAddress() + "->"
            + groupMembers.toString() + ")");

      // Send the query to everybody including us
      RspList responses = dvdb.getDispatcher().castMessage(groupMembers,
          CJDBCGroupMessage.getMessage(new Rollback(transactionId)),
          GroupRequest.GET_ALL, this.rollbackTimeout);

      if (logger.isDebugEnabled())
        logger
            .debug("rollback of transaction " + transactionId + " completed.");

      if (responses.numSuspectedMembers() > 0)
      { // Some controllers failed ... too bad !
        logger
            .warn(responses.numSuspectedMembers()
                + " controller(s) died during execution of rollback for transaction "
                + transactionId);
      }

      // List of controllers that gave a AllBackendsFailedException
      Vector failedOnAllBackends = null;
      SQLException exception = null;
      int size = groupMembers.size();
      boolean success = false;
      // Get the result of each controller
      for (int i = 0; i < size; i++)
      {
        Address address = (Address) groupMembers.get(i);
        if (responses.isSuspected(address))
        {
          logger.warn("Controller " + address + " is suspected of failure.");
          continue;
        }
        Object r = responses.get(address);
        if (r instanceof Boolean)
        {
          if (((Boolean) r).booleanValue())
            success = true;
          else
            logger.error("Unexpected result for controller  " + address);
        }
        else if (r instanceof AllBackendsFailedException)
        {
          if (failedOnAllBackends == null)
            failedOnAllBackends = new Vector();
          failedOnAllBackends.add(address);
          if (logger.isDebugEnabled())
            logger.debug("rollback failed on all backends of controller "
                + address + " (" + r + ")");
        }
        else if (r instanceof SQLException)
        {
          String msg = "rollback of transaction " + transactionId
              + " failed on controller " + address + " (" + r + ")";
          logger.warn(msg);
          exception = (SQLException) r;
        }
      }

      if (failedOnAllBackends != null)
      { // Notify all controllers where all backend failed that the rollback
        // completed with 'success'
        AbstractRequest request = new UnknownRequest("rollback", false, 0, "\n");
        request.setTransactionId(transactionId);
        dvdb.getDispatcher().castMessage(
            failedOnAllBackends,
            CJDBCGroupMessage
                .getMessage(new NotifyCompletion(request, success)),
            GroupRequest.GET_NONE, rollbackTimeout);
      }

      if (success)
        return; // This is a success if at least one controller has succeeded

      if (exception != null)
        throw exception;

      // At this point, all controllers failed
      String msg = "Transaction " + transactionId
          + " failed to rollback on all controllers";
      logger.warn(msg);
      throw new SQLException(msg);
    }
    catch (SQLException e)
    {
      String msg = "Transaction " + transactionId + " rollback failed (" + e
          + ")";
      logger.warn(msg);
      throw e;
    }
  }

}

---