SingleDBDistributedRequestManager.java

説明を見る。

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

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.AbstractWriteRequest;
import org.objectweb.cjdbc.common.sql.StoredProcedure;
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.CacheInvalidate;

public class SingleDBDistributedRequestManager
    extends
      DistributedRequestManager
{

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

  //
  // Request handling
  //

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

      Vector allButUs = dvdb.getChannel().getView().getMembers();
      allButUs.remove(dvdb.getChannel().getLocalAddress());
      // TODO: Exclude controllers without cache from the invalidation list

      if (logger.isDebugEnabled())
        logger
            .debug("Sending cache invalidation to all controllers for request "
                + request.getSQLShortForm(dvdb.getSQLShortFormLength()));
      // Send the query to everybody except us
      RspList responses = dvdb.getDispatcher().castMessage(allButUs,
          CJDBCGroupMessage.getMessage(new CacheInvalidate(request)),
          GroupRequest.GET_ALL, request.getTimeout());

      // Execute locally
      try
      {
        scheduleExecWriteRequest((AbstractWriteRequest) request);
        execWriteRequestResult = loadBalanceExecWriteRequest((AbstractWriteRequest) request);
      }
      catch (SQLException e)
      {
        throw e;
      }
      catch (RuntimeException re)
      {
        logger.warn("Error while executing distributed write request", re);
        throw new SQLException(re.getMessage());
      }
      catch (AllBackendsFailedException abe)
      {
        throw new SQLException(abe.getMessage());
      }
      finally
      {
        updateAndNotifyExecWriteRequest((AbstractWriteRequest) request, true);
      }
      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.getSQLShortForm(dvdb.getSQLShortFormLength()));
      }

      SQLException exception = null;
      Vector suspectedMembers = responses.getSuspectedMembers();
      int size = suspectedMembers.size();
      // Get the result of each controller
      for (int i = 0; i < size; i++)
      {
        Address address = (Address) suspectedMembers.get(i);
        logger.warn("Controller " + address + " is suspected of failure.");
        continue;
      }

      return execWriteRequestResult;
    }
    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 allButUs = dvdb.getChannel().getView().getMembers();
      allButUs.remove(dvdb.getChannel().getLocalAddress());
      // TODO: Exclude controllers without cache from the invalidation list

      if (logger.isDebugEnabled())
        logger
            .debug("Sending cache invalidation to all controllers for request "
                + request.getSQLShortForm(dvdb.getSQLShortFormLength()));
      // Send the query to everybody except us
      RspList responses = dvdb.getDispatcher().castMessage(allButUs,
          CJDBCGroupMessage.getMessage(new CacheInvalidate(request)),
          GroupRequest.GET_ALL, request.getTimeout());

      // Execute locally
      try
      {
        scheduleExecWriteRequest((AbstractWriteRequest) request);
        execWriteRequestResult = loadBalanceExecWriteRequestWithKeys((AbstractWriteRequest) request);
      }
      catch (SQLException e)
      {
        throw e;
      }
      catch (RuntimeException re)
      {
        logger.warn("Error while executing distributed write request", re);
        throw new SQLException(re.getMessage());
      }
      catch (AllBackendsFailedException abe)
      {
        throw new SQLException(abe.getMessage());
      }
      finally
      {
        updateAndNotifyExecWriteRequest((AbstractWriteRequest) request, true);
      }
      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.getSQLShortForm(dvdb.getSQLShortFormLength()));
      }

      SQLException exception = null;
      Vector suspectedMembers = responses.getSuspectedMembers();
      int size = suspectedMembers.size();
      // Get the result of each controller
      for (int i = 0; i < size; i++)
      {
        Address address = (Address) suspectedMembers.get(i);
        logger.warn("Controller " + address + " is suspected of failure.");
        continue;
      }

      return execWriteRequestResult;
    }
    catch (SQLException e)
    {
      String msg = "Request '" + request + "' failed (" + e + ")";
      logger.warn(msg);
      throw e;
    }
  }

  public int execDistributedWriteStoredProcedure(StoredProcedure proc)
      throws SQLException
  {
    // TODO Auto-generated method stub
    return 0;
  }

  public void distributedCommit(long transactionId) throws SQLException
  {
    // There is a little subtlety here. We call
    // DistributedRequestManager.commit() but as the transaction has already
    // been removed from the writeTransaction list, it will execute as a local
    // commit.
    super.commit(transactionId);
  }

  public void distributedRollback(long transactionId) throws SQLException
  {
    // There is a little subtlety here. We call
    // DistributedRequestManager.rollback() but as the transaction has already
    // been removed from the writeTransaction list, it will execute as a local
    // rollback.
    super.rollback(transactionId);
  }
}

---