src/org/objectweb/cjdbc/controller/virtualdatabase/DistributedVirtualDatabase.java

説明を見る。

package org.objectweb.cjdbc.controller.virtualdatabase;

import java.io.Serializable;
import java.net.URL;
import java.sql.SQLException;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.Hashtable;
import java.util.Iterator;
import java.util.Vector;
import java.util.Map.Entry;

import javax.management.NotCompliantMBeanException;

import org.dom4j.DocumentException;
import org.jgroups.Address;
import org.jgroups.JChannel;
import org.jgroups.MembershipListener;
import org.jgroups.Message;
import org.jgroups.MessageListener;
import org.jgroups.View;
import org.jgroups.blocks.GroupRequest;
import org.jgroups.blocks.MessageDispatcher;
import org.jgroups.blocks.RequestHandler;
import org.jgroups.util.RspList;
import org.objectweb.cjdbc.common.exceptions.ControllerException;
import org.objectweb.cjdbc.common.exceptions.VirtualDatabaseException;
import org.objectweb.cjdbc.common.i18n.Translate;
import org.objectweb.cjdbc.common.jmx.JmxException;
import org.objectweb.cjdbc.common.jmx.notifications.CjdbcNotificationList;
import org.objectweb.cjdbc.common.shared.BackendInfo;
import org.objectweb.cjdbc.common.sql.filters.AbstractBlobFilter;
import org.objectweb.cjdbc.common.xml.DatabasesXmlTags;
import org.objectweb.cjdbc.controller.authentication.AuthenticationManager;
import org.objectweb.cjdbc.controller.backend.BackendRecoveryPolicy;
import org.objectweb.cjdbc.controller.backend.DatabaseBackend;
import org.objectweb.cjdbc.controller.core.Controller;
import org.objectweb.cjdbc.controller.jmx.MBeanServerManager;
import org.objectweb.cjdbc.controller.jmx.RmiConnector;
import org.objectweb.cjdbc.controller.requestmanager.RequestManager;
import org.objectweb.cjdbc.controller.requestmanager.distributed.DistributedRequestManager;
import org.objectweb.cjdbc.controller.virtualdatabase.protocol.BackendStatus;
import org.objectweb.cjdbc.controller.virtualdatabase.protocol.BackendTransfer;
import org.objectweb.cjdbc.controller.virtualdatabase.protocol.CJDBCGroupMessage;
import org.objectweb.cjdbc.controller.virtualdatabase.protocol.Commit;
import org.objectweb.cjdbc.controller.virtualdatabase.protocol.ControllerName;
import org.objectweb.cjdbc.controller.virtualdatabase.protocol.DistributedRequest;
import org.objectweb.cjdbc.controller.virtualdatabase.protocol.EnableBackend;
import org.objectweb.cjdbc.controller.virtualdatabase.protocol.Rollback;
import org.objectweb.cjdbc.controller.virtualdatabase.protocol.VirtualDatabaseConfiguration;

public class DistributedVirtualDatabase extends VirtualDatabase
    implements
      RequestHandler,
      MessageListener,
      MembershipListener
{
  //
  // How the code is organized ?
  //
  // 1. Member variables
  // 2. Constructor(s)
  // 3. Request handling
  // 4. Transaction handling
  // 5. Database backend management
  // 6. Distribution management (multicast)
  // 7. Getter/Setter (possibly in alphabetical order)
  //

  // Distribution

  private String            groupName      = null;
  private Hashtable         controllersMap;
  private HashMap           backendRecoveryPolicy;
  private Hashtable         backendsPerController;

  private JChannel          channel        = null;
  private MessageDispatcher dispatcher;
  private View              currentView    = null;
  private Vector            allMemberButUs = null;

  //
  // Constructors
  //

  public DistributedVirtualDatabase(Controller controller, String name,
      String groupName, int maxConnections, boolean pool, int minThreads,
      int maxThreads, long maxThreadIdleTime, int sqlShortFormLength,
      AbstractBlobFilter blobFilter) throws NotCompliantMBeanException,
      JmxException
  {
    super(controller, name, maxConnections, pool, minThreads, maxThreads,
        maxThreadIdleTime, sqlShortFormLength, blobFilter);

    this.groupName = groupName;
    backendRecoveryPolicy = new HashMap();
    backendsPerController = new Hashtable();
    controllersMap = new Hashtable();
  }

  protected void finalize() throws Throwable
  {
    quitChannel();
    super.finalize();
  }

  public void addBackend(DatabaseBackend db) throws VirtualDatabaseException
  {
    // Add the backend to the virtual database.
    super.addBackend(db);

    // Send a group message if already joined group
    if(allMemberButUs!=null)
      broadcastBackendInformation(allMemberButUs);
  }

  public void quitChannel()
  {
    if (channel != null)
    {
      channel.close();
    }
  }

  public String getControllerName()
  {
    return controller.getName();
  }

  public String getGroupName()
  {
    return groupName;
  }

  public void setGroupName(String groupName)
  {
    this.groupName = groupName;
  }

  public void addBackendRecoveryPolicy(BackendRecoveryPolicy policy)
  {
    backendRecoveryPolicy.put(policy.getBackendName(), policy);
  }

  public void setRequestManager(RequestManager requestManager)
  {
    if (!(requestManager instanceof DistributedRequestManager))
      throw new RuntimeException(
          "A distributed virtual database can only work with a distributed request manager.");
    this.requestManager = requestManager;
  }

  //
  //  Distribution handling
  //

  public void joinGroup() throws Exception
  {
    try
    {
      // Read the protocol stack configuration from jgroups.xml
      URL jgroupConfigFile = DistributedVirtualDatabase.class
          .getResource("/jgroups.xml");
      if (jgroupConfigFile == null)
        logger.warn(Translate
            .get("virtualdatabase.distributed.jgroups.xml.not.found"));
      else
        logger.info(Translate.get("virtualdatabase.distributed.jgroups.using",
            jgroupConfigFile.toString()));
      channel = new JChannel(jgroupConfigFile);
      if (logger.isDebugEnabled())
        logger.debug("Group communication channel is configured as follows: "
            + channel.getProperties());

      // Join the group
      channel.connect(groupName);
      dispatcher = new MessageDispatcher(channel /*
       * group
       * channel
       */
      , this /* messageListener */
      , this /* membershipListener */
      , this /* requestHandler */
      , false /* deadlock detection */
      );

      logger.info("Group " + groupName + " connected to "
          + channel.getLocalAddress());

      // Add ourselves to the list of controllers
      controllersMap.put(channel.getLocalAddress(), controller.getJmxName());

      // Check if we are alone or not
      currentView = channel.getView();
      if (currentView.size() == 1)
      {
        logger.info(Translate.get(
            "virtualdatabase.distributed.configuration.first.in.group",
            groupName));
        allMemberButUs = new Vector(1);
        return;
      }

      // Compute the group list without us
      allMemberButUs = (Vector) currentView.getMembers().clone();

      // Our controller id is our position in the membership.
      // This assumes that all controllers have the view in the same order.
      ((DistributedRequestManager) requestManager)
          .setControllerId(allMemberButUs.indexOf(channel.getLocalAddress()));
      allMemberButUs.remove(channel.getLocalAddress());

      // Check with the other controller that our config is compatible
      if (checkConfigurationCompatibility(allMemberButUs) == false)
      {
        String msg = Translate
            .get("virtualdatabase.distributed.configuration.not.compatible");
        logger.error(msg);
        throw new ControllerException(msg);
      }

      logger.info(Translate
          .get("virtualdatabase.distributed.configuration.compatible"));

      // Distribute backends among controllers knowing that at this point
      // there is no conflict on the backend distribution policies.
      broadcastBackendInformation(allMemberButUs);
    }
    catch (Exception e)
    {
      String msg = Translate.get("virtualdatabase.distributed.joingroup.error",
          groupName);
      if (e instanceof RuntimeException)
        logger.error(msg, e);
      throw new Exception(msg + " (" + e + ")");
    }
  }

  public JChannel getChannel()
  {
    return channel;
  }

  public MessageDispatcher getDispatcher()
  {
    return dispatcher;
  }

  public View getCurrentView()
  {
    return currentView;
  }

  public Vector getAllMemberButUs()
  {
    return allMemberButUs;
  }

  private boolean checkConfigurationCompatibility(Vector dest)
  {
    if (logger.isDebugEnabled())
      logger.debug(Translate
          .get("virtualdatabase.distributed.configuration.checking"));

    // Send our configuration
    RspList rspList = dispatcher.castMessage(dest, CJDBCGroupMessage
        .getMessage(new VirtualDatabaseConfiguration(this)),
        GroupRequest.GET_ALL, CJDBCGroupMessage.defaultCastTimeOut);

    // Check that everybody agreed
    Vector results = rspList.getResults();
    int size = results.size();
    if (size == 0)
      logger.warn(Translate
          .get("virtualdatabase.distributed.configuration.checking.noanswer"));
    for (int i = 0; i < size; i++)
    {
      Object response = results.get(i);
      if (response instanceof Boolean)
      {
        if (((Boolean) response).booleanValue() == false)
          return false;
      }
      else
      {
        logger
            .error("Unexpected response while checking configuration compatibility: "
                + response);
        return false;
      }
    }

    // Ok, everybody replied true (config is compatible)
    return true;
  }

  private void broadcastBackendInformation(Vector dest)
  {
    logger
        .debug(Translate
            .get("virtualdatabase.distributed.configuration.querying.remote.status"));

    // Send our backend status
    RspList rspList = dispatcher.castMessage(dest, CJDBCGroupMessage
        .getMessage(new BackendStatus(getBackendsInfo(backends))),
        GroupRequest.GET_ALL, CJDBCGroupMessage.defaultCastTimeOut);

    int size = dest.size();
    for (int i = 0; i < size; i++)
    {
      // Add the backend configuration of every remote controller
      Address addr = (Address) dest.get(i);
      if (rspList.isReceived(addr))
      {
        if (logger.isDebugEnabled())
          logger
              .debug(Translate
                  .get(
                      "virtualdatabase.distributed.configuration.updating.backend.list",
                      addr.toString()));
      }
      else
        logger.warn(Translate.get(
            "virtualdatabase.distributed.unable.get.remote.status", addr
                .toString()));
    }
  }

  public boolean isCompatibleBackend(BackendInfo backend)
      throws VirtualDatabaseException
  {
    try
    {
      acquireReadLockBackendLists();
    }
    catch (InterruptedException e)
    {
      String msg = "Unable to acquire read lock on backend list in isCompatibleBackend ("
          + e + ")";
      logger.error(msg);
      throw new VirtualDatabaseException(msg);
    }

    try
    {
      // Find the backend
      String backendURL = backend.getUrl();
      int size = backends.size();
      DatabaseBackend b = null;
      for (int i = 0; i < size; i++)
      {
        b = (DatabaseBackend) backends.get(i);
        if (b.getURL().equals(backendURL))
          return false;
      }
    }
    catch (RuntimeException re)
    {
      throw new VirtualDatabaseException(re);
    }
    finally
    {
      releaseReadLockBackendLists();
    }
    // This backend does not exist here
    return true;
  }

  //
  // Message dispatcher request handling
  //

  public Object handle(Message msg)
  {
    try
    {
      if (msg != null)
      {
        Object content = msg.getObject();
        if (content instanceof DistributedRequest)
        { //Distributed request execution
          if (logger.isDebugEnabled())
            logger.debug(getControllerName() + ": DistributedRequest "
                + ((DistributedRequest) content).getRequest().getId()
                + " from " + controllersMap.get(msg.getSrc()));
          return ((Serializable) ((DistributedRequest) content)
              .executeRequest((DistributedRequestManager) this.requestManager));
        }
        if (content instanceof BackendTransfer)
        {
          logger.info(getControllerName()+":Received transfer command");
          BackendTransfer transfer = (BackendTransfer) content;
          BackendInfo info = transfer.getInfo();
          DatabaseBackend backend = new DatabaseBackend(info);
          try
          {
            this.addBackend(backend);
          }
          catch (Exception e1)
          {
            logger.error("Transfer failed",e1);
            return e1;
          }
          logger.info(getControllerName()+":Enable backend");
          enableBackend(backend.getName());
          return Boolean.TRUE;
        }
        if (content instanceof ControllerName)
        {
          String controllerName = ((ControllerName) content).getJmxName();
          if (logger.isDebugEnabled())
            logger.debug(getControllerName() + ": New ControllerName "
                + controllerName);
          controllersMap.put(msg.getSrc(), controllerName);
          return Boolean.TRUE;
        }
        if (content instanceof Commit)
        { //Distributed commit execution
          if (logger.isDebugEnabled())
            logger.debug(getControllerName() + ": Commit from "
                + controllersMap.get(msg.getSrc()));
          return ((Serializable) ((Commit) content)
              .commit((DistributedRequestManager) this.requestManager));
        }
        if (content instanceof Rollback)
        { //Distributed commit execution
          if (logger.isDebugEnabled())
            logger.debug(getControllerName() + ": Rollback from "
                + controllersMap.get(msg.getSrc()));
          return ((Serializable) ((Rollback) content)
              .rollback((DistributedRequestManager) this.requestManager));
        }
        else if (content instanceof VirtualDatabaseConfiguration)
        { // Check if given configuration is compatible with the local one
          VirtualDatabaseConfiguration vdc = (VirtualDatabaseConfiguration) content;
          // Send notification
          if (MBeanServerManager.isJmxEnabled())
          {
            Hashtable data = new Hashtable();
            data.put("controllerName", vdc.getControllerName());
            data.put("rmiconnector", new String[]{vdc.getRmiHostname(),
                vdc.getRmiPort()});
            RmiConnector.broadcastNotification(this,
                CjdbcNotificationList.DISTRIBUTED_CONTROLLER_ADDED,
                CjdbcNotificationList.NOTIFICATION_LEVEL_INFO, Translate
                    .get("notification.distributed.controller.added", this
                        .getName()), data);
          }
          controllersMap.put(msg.getSrc(), vdc.getControllerJmxName());
          if (logger.isDebugEnabled())
            logger.debug("VirtualDatabaseConfiguration from "
                + vdc.getControllerName());

          // Send controller name to new comer
          Vector target = new Vector();
          target.add(msg.getSrc());
          dispatcher.castMessage(target, CJDBCGroupMessage
              .getMessage(new ControllerName(controller.getName(), controller
                  .getJmxName())), GroupRequest.GET_ALL,
              CJDBCGroupMessage.defaultCastTimeOut);

          // Broadcast backends
          dispatcher.castMessage(target, CJDBCGroupMessage
              .getMessage(new BackendStatus(getBackendsInfo(backends))),
              GroupRequest.GET_ALL, CJDBCGroupMessage.defaultCastTimeOut);

          if (isLocalMessage(msg))
            return Boolean.TRUE;
          else
            return new Boolean((vdc.isCompatible(this)));
        }
        else if (content instanceof BackendStatus)
        { // Update backend list from sender
          Address src = msg.getSrc();
          ArrayList remoteBackendList = ((BackendStatus) content).getBackends();
          backendsPerController.put(src, remoteBackendList);
          if (logger.isInfoEnabled())
            logger
                .info(Translate
                    .get(
                        "virtualdatabase.distributed.configuration.updating.backend.list",
                        controllersMap.get(src)));
        }
        else if (content instanceof EnableBackend)
        {
          Address src = msg.getSrc();
          ArrayList remoteBackendList = (ArrayList) backendsPerController
              .get(src);
          DatabaseBackend enabledBackend = ((EnableBackend) content)
              .getBackend();
          int size = remoteBackendList.size();
          boolean backendFound = false;
          for (int i = 0; i < size; i++)
          {
            DatabaseBackend b = (DatabaseBackend) remoteBackendList.get(i);
            if (b.equals(enabledBackend))
            {
              logger.info("Backend " + b.getName() + " enabled on controller "
                  + controllersMap.get(src));
              remoteBackendList.set(i, enabledBackend);
              backendFound = true;
              break;
            }
          }
          if (!backendFound)
          {
            logger.warn("Updating backend list with unknown backend "
                + enabledBackend.getName() + " enabled on controller "
                + controllersMap.get(src));
            remoteBackendList.add(enabledBackend);
          }
        }
        else
        {
          logger.warn("Unhandled message type received: " + content.getClass()
              + "(" + content + ")");
        }
      }
      else
      {
        String errorMsg = "Invalid null message";
        logger.error(errorMsg);
        return new ControllerException(errorMsg);
      }
      return msg.makeReply();
    }
    catch (Exception e)
    {
      if (e instanceof RuntimeException)
        logger.warn("Error while handling group message:"
            + msg.getObject().getClass(), e);
      return e;
    }
  }

  private boolean isLocalMessage(Message msg)
  {
    return channel.getLocalAddress().equals(msg.getSrc());
  }

  public void getBackendStatus()
  {
    if (logger.isDebugEnabled())
      logger.debug("Requesting remote controllers status");
    RspList rspList = dispatcher.castMessage(null, CJDBCGroupMessage
        .getMessage(new BackendStatus(getBackendsInfo(backends))),
        GroupRequest.GET_ALL, CJDBCGroupMessage.defaultCastTimeOut);
    Vector results = rspList.getResults();
    int size = results.size();
    for (int i = 0; i < size; i++)
    {
      ArrayList b = (ArrayList) results.get(i);
      int bSize = b.size();
      if (bSize == 0)
        logger.debug("No Database Backends");
      else
        for (int j = 0; j < bSize; j++)
          logger.debug(((DatabaseBackend) b.get(j)).getXml());
    }
  }

  //
  // Message listener interface
  //

  public void receive(Message msg)
  {
    if (logger.isDebugEnabled())
      logger.debug("Received message: " + msg + " (header:"
          + msg.printObjectHeaders() + ")");
  }

  public byte[] getState()
  {
    return null;
  }

  public void setState(byte[] state)
  {
  }

  //
  // Membership support
  //

  public void block()
  {
    if (logger.isDebugEnabled())
      logger.debug("Messages should be freezed until new view is installed.");
  }

  public void suspect(Address suspectedMember)
  {
    logger.warn("Member " + suspectedMember + " is suspected of failure");
  }

  public void viewAccepted(View newView)
  {
    Vector newMembers = newView.getMembers();
    int newSize = newMembers.size();
    for (int i = 0; i < newSize; i++)
    {
      Address addr = (Address) newMembers.get(i);
      if (!currentView.containsMember(addr))
      {
        logger.info(controller.getName() + ":New controller " + addr.toString()
            + " detected.");
      }
    }
    Vector currentMembers = currentView.getMembers();
    logger.info(controller.getName() + ":Current Members:" + currentMembers);
    int currentSize = currentMembers.size();
    for (int i = 0; i < currentSize; i++)
    {
      Address addr = (Address) currentMembers.get(i);
      if (!newView.containsMember(addr))
      {
        logger.warn("Controller " + addr + " has failed.");
      }
    }
    currentView = newView;
    currentMembers = currentView.getMembers();
    logger.info(controller.getName() + ":new Members:" + currentMembers);
    allMemberButUs = ((Vector) currentView.getMembers().clone());
    allMemberButUs.remove(channel.getLocalAddress());
  }

  //
  // Getter/Setter and tools (equals, ...)
  //

  public HashMap getBackendRecoveryPolicy()
  {
    return backendRecoveryPolicy;
  }

  public boolean isDistributed()
  {
    return true;
  }

  public boolean equals(Object other)
  {
    if ((other == null)
        || (!(other instanceof org.objectweb.cjdbc.controller.virtualdatabase.DistributedVirtualDatabase)))
      return false;
    else
    {
      DistributedVirtualDatabase db = (org.objectweb.cjdbc.controller.virtualdatabase.DistributedVirtualDatabase) other;
      return name.equals(db.getDatabaseName())
          && groupName.equals(db.getGroupName());
    }
  }

  protected String getDistributionXml()
  {
    StringBuffer info = new StringBuffer();
    info.append("<" + DatabasesXmlTags.ELT_Distribution + " "
        + DatabasesXmlTags.ATT_groupName + "=\"" + groupName + "\">");

    for (Iterator iter = backendRecoveryPolicy.values().iterator(); iter
        .hasNext();)
    {
      BackendRecoveryPolicy p = (BackendRecoveryPolicy) iter.next();
      info.append(p.getXml());
    }

    info.append("</" + DatabasesXmlTags.ELT_Distribution + ">");
    return info.toString();
  }

  public String[] viewControllerList()
  {
    if (logger.isInfoEnabled())
    {
      logger.info(channel.getLocalAddress() + " see members:"
          + currentView.getMembers() + " and has mapping:" + controllersMap);
    }
    String[] members = new String[controllersMap.keySet().size()];
    Iterator iter = controllersMap.keySet().iterator();
    int i = 0;
    while (iter.hasNext())
    {
      members[i] = (String) controllersMap.get(iter.next());
      i++;
    }
    return members;
  }

  public void setAuthenticationManager(
      AuthenticationManager authenticationManager)
  {
    super.setAuthenticationManager(authenticationManager);
  }

  public Hashtable viewGroupBackends() throws VirtualDatabaseException
  {
    Hashtable map = new Hashtable(controllersMap.size());
    Iterator iter = backendsPerController.keySet().iterator();
    Address addr;
    while (iter.hasNext())
    {
      addr = (Address) iter.next();
      map.put(controllersMap.get(addr), backendsPerController.get(addr));
    }
    return map;
  }

  public ArrayList getBackendsInfo(ArrayList backendsObject)
  {
    int size = backendsObject.size();
    ArrayList infos = new ArrayList(size);
    DatabaseBackend backend;
    for (int i = 0; i < size; i++)
    {
      backend = (DatabaseBackend) backendsObject.get(i);
      infos.add(createBackendInfo(backend,false));
    }
    return infos;
  }

  public BackendInfo createBackendInfo(DatabaseBackend backend, boolean useXml)
  {
    if (useXml)
      try
      {
        return new BackendInfo(backend.getXml());
      }
      catch (DocumentException e)
      {
        // go to next method. cause we haven't returned yet
      }
    return new BackendInfo(backend.getName(), backend.getDriverPath(), backend
        .getDriverClassName(), backend.getURL(), backend
        .getConnectionTestStatement());
  }

  public void removeBackend(String backend) throws VirtualDatabaseException
  {
    super.removeBackend(backend);

    // Send a group message to update backend list
    broadcastBackendInformation(allMemberButUs);
  }

  public void transferBackend(String backend, String controllerDestination)
      throws VirtualDatabaseException
  {
    // Get the target controller
    Vector dest = new Vector(1);
    Iterator iter = controllersMap.entrySet().iterator();
    Entry entry;
    Address targetAddr = null;
    while (iter.hasNext())
    {
      entry = (Entry) iter.next();
      if (entry.getValue().equals(controllerDestination))
        targetAddr = (Address) entry.getKey();
    }
    if (targetAddr == null)
      throw new VirtualDatabaseException("Cannot find controller:"
          + controllerDestination + " in group");
    dest.add(targetAddr);

    // Get reference on backend
    DatabaseBackend db = getAndCheckBackend(backend, NO_CHECK_BACKEND);
    
    //  Disable backend
    try
    {
      DistributedRequestManager manager =((DistributedRequestManager) requestManager) ;
      if(hasRecoveryLog())
        manager.disableBackendForCheckpoint(db, "transfer of " + backend + " to :" + controllerDestination);
      else
        manager.disableBackend(db);
    }
    catch (SQLException e)
    {
      throw new VirtualDatabaseException(e.getMessage());
    }
    
    //  Send backend transfer message
    logger.info("Sending transfer message to:"+targetAddr);
    dispatcher.castMessage(dest, CJDBCGroupMessage
        .getMessage(new BackendTransfer(controllerDestination,createBackendInfo(db,true))),
        GroupRequest.GET_ALL, CJDBCGroupMessage.defaultCastTimeOut);    
    
    // Remove backend from this controller
    removeBackend(db);
    
    // Broadcast updated backend list
    broadcastBackendInformation(allMemberButUs);
  }
}

---