src/org/objectweb/cjdbc/controller/loadbalancer/BackendWorkerThread.java

説明を見る。

package org.objectweb.cjdbc.controller.loadbalancer;

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

import org.objectweb.cjdbc.common.i18n.Translate;
import org.objectweb.cjdbc.common.log.Trace;
import org.objectweb.cjdbc.controller.backend.DatabaseBackend;
import org.objectweb.cjdbc.controller.loadbalancer.tasks.AbstractTask;

public class BackendWorkerThread extends Thread
{
  //
  // How the code is organized ?
  // 1. Member variables
  // 2. Constructor(s)
  // 3. Task management
  // 4. Getter/Setters
  //

  private AbstractLoadBalancer loadBalancer;
  private DatabaseBackend      backend;
  private ArrayList            taskList;
  private ArrayList            tidList;
  private boolean              isKilled            = false;

  // true if we are currently processing a task
  private boolean              currentlyProcessing = false;

  private Trace                logger              = null;

  /*
   * Constructor
   */

  public BackendWorkerThread(DatabaseBackend backend,
      AbstractLoadBalancer loadBalancer) throws SQLException
  {
    this("BackendWorkerThread for backend:" + backend.getName()
        + " and a loadBalancer level:" + loadBalancer.getRAIDbLevel(), backend,
        loadBalancer);
  }

  public BackendWorkerThread(String name, DatabaseBackend backend,
      AbstractLoadBalancer loadBalancer) throws SQLException
  {
    super(name);
    // Sanity checks
    if (backend == null)
    {
      String msg = Translate.get("backendworkerthread.null.backend");
      logger = Trace
          .getLogger("org.objectweb.cjdbc.controller.backend.DatabaseBackend");
      logger.error(msg);
      throw new SQLException(msg);
    }

    backend.checkDriverCompliance();

    logger = Trace
        .getLogger("org.objectweb.cjdbc.controller.backend.DatabaseBackend."
            + backend.getName());

    if (loadBalancer == null)
    {
      String msg = Translate.get("backendworkerthread.null.loadbalancer");
      logger.error(msg);
      throw new SQLException(msg);
    }

    this.backend = backend;
    this.loadBalancer = loadBalancer;
    taskList = new ArrayList();
  }

  /*
   * Task Management
   */

  public void addTask(AbstractTask task)
  {
    if (!isKilled)
    {
      taskList.add(task);
      // We assume that all requests here are writes
      backend.addPendingWriteRequest(task);
    }
    else
      task.notifyCompletion();
  }

  public void addTask(AbstractTask task, long transactionId)
  {
    if (!isKilled)
    {
      task.setHasTid(true);
      addTask(task);
      if (tidList == null) // Instantiate the list if needed
        tidList = new ArrayList();
      tidList.add(new Long(transactionId));
    }
    else
      task.notifyCompletion();
  }

  public void addPriorityTask(AbstractTask task)
  {
    if (!isKilled)
    {
      taskList.add(0, task);
      // We assume that all requests here are writes
      backend.addPendingWriteRequest(task);
    }
    else
      task.notifyCompletion();
  }

  public void addPriorityTask(AbstractTask task, long transactionId)
  {
    if (!isKilled)
    {
      task.setHasTid(true);
      addPriorityTask(task);
      tidList.add(0, new Long(transactionId));
    }
    else
      task.notifyCompletion();
  }

  public boolean removeTask(AbstractTask task)
  {
    backend.removePendingRequest(task);
    if (tidList == null)
      return taskList.remove(task);
    else
    {
      int idx = taskList.indexOf(task);
      if (idx == -1)
        return false;
      else
      {
        taskList.remove(idx);
        tidList.remove(idx);
        return true;
      }
    }
  }

  public void waitForAllTasksToComplete(long transactionId)
  {
    if ((transactionId == 0) || (tidList == null))
      return;

    Long tid = new Long(transactionId);
    synchronized (this)
    {
      if (!tidList.contains(tid))
      {
        if (currentlyProcessing)
        { // We are not sure that the task currently executing belongs to the
          // transaction we are interested in or not. So we wait for the
          // completion of the current task, it is safer.
          try
          {
            if (logger.isDebugEnabled())
              logger.debug(Translate.get("backendworkerthread.waiting.task"));
            wait();
          }
          catch (InterruptedException ignore)
          {
          }
          return;
        }
        else
          return;
      }

      while (tidList.contains(tid))
      {
        if (logger.isDebugEnabled())
          logger.debug(Translate.get("backendworkerthread.waiting.transaction",
              String.valueOf(tid)));

        try
        {
          wait();
        }
        catch (InterruptedException ignore)
        {
        }
      }
    }
  }

  public void waitForAllTasksToComplete()
  {
    synchronized (this)
    {
      Object current;
      if (taskList.size() == 0)
      {
        if (currentlyProcessing)
        {
          try
          {
            if (logger.isDebugEnabled())
              logger.debug(Translate.get("backendworkerthread.waiting.task"));
            wait();
          }
          catch (InterruptedException ignore)
          {
            logger.warn(Translate
                .get("backendworkerthread.no.full.task.synchronization"));
          }
          return;
        }
        else
        { // No task currently executing
          return;
        }
      }
      else
        current = taskList.get(taskList.size() - 1);

      if (logger.isDebugEnabled())
        logger.debug(Translate.get("backendworkerthread.waiting.request",
            current.toString()));

      while (taskList.contains(current))
      {
        try
        {
          wait();
        }
        catch (InterruptedException ignore)
        {
        }
      }
    }
  }

  public synchronized void kill()
  {
    if (backend.isKilled())
      return;

    String msg = "Backend " + backend.getName() + " is shutting down";
    // Remove all tasks
    while (!taskList.isEmpty())
    {
      AbstractTask task = (AbstractTask) taskList.remove(0);
      try
      {
        task.notifyFailure(this, 1, new SQLException(msg));
      }
      catch (SQLException ignore)
      {
      }
    }
    isKilled = true;
    notify(); // Wake up thread
    logger.info(msg);
    try
    {
      // This ensure that all worker threads get removed from the load balancer
      // list and that the backend state is set to disable.
      loadBalancer.disableBackend(backend);
    }
    catch (SQLException ignore)
    {
    }
  }

  public void run()
  {
    AbstractTask task = null;

    while (!isKilled)
    {
      synchronized (this)
      {
        if (taskList.isEmpty())
        { // Nothing to do, go to bed!
          try
          {
            wait();
          }
          catch (InterruptedException e)
          {
            logger.warn(Translate.get("backendworkerthread.wait.interrupted"));
          }
        }
        try
        { // Take the 1st task from the list
          task = (AbstractTask) taskList.remove(0);
          currentlyProcessing = true;
        }
        catch (IndexOutOfBoundsException oob)
        {
          logger.warn(Translate.get("backendworkerthread.no.task"));
        }
      }
      // Execute the task out of the sync block
      try
      {
        if (logger.isDebugEnabled())
          logger.debug(Translate.get("backendworkerthread.execute.task", task
              .toString()));
        task.execute(this);
      }
      catch (SQLException e)
      {
        // Task should have notified of failure
        logger.warn(Translate.get("backendworkerthread.task.failed", e));
      }
      catch (RuntimeException re)
      {
        // We can't know for sure if the task has notified the failure or not.
        // To prevent a deadlock, we force the failure notification here.
        try
        {
          task.notifyFailure(this, 1, new SQLException(re.getMessage()));
        }
        catch (SQLException e1)
        {
          // just notify
        }
        logger.fatal(Translate.get(
            "backendworkerthread.task.runtime.exception", task.toString()), re);
      }
      finally
      {
        try
        {
          backend.removePendingRequest(task);
        }
        catch (RuntimeException e)
        {
          logger.warn(
              Translate.get("backendworkerthread.remove.task.error", e), e);
        }
      }

      if (task.hasTid())
      {
        // Notify the completion of the task if someone is waiting for
        // the completion of this transaction (@see #waitForAllTasksToComplete)
        synchronized (this)
        {
          if (tidList != null)
            tidList.remove(0);
          notifyAll();
          currentlyProcessing = false;
        }
      }
    }

    // Automatically disable the backend when the thread dies
    try
    {
      if (backend.isReadEnabled() || backend.isWriteEnabled())
        loadBalancer.disableBackend(backend);
    }
    catch (SQLException e)
    {
      logger.error(Translate.get("backendworkerthread.backend.disable.failed",
          new String[]{backend.getName(), e.getMessage()}));
    }
  }

  /*
   * Getter/Setter
   */

  public DatabaseBackend getBackend()
  {
    return backend;
  }

  public Trace getLogger()
  {
    return logger;
  }
}

---