src/org/objectweb/cjdbc/controller/recoverylog/JDBCRecoverThread.java

説明を見る。

package org.objectweb.cjdbc.controller.recoverylog;

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.AbstractLoadBalancer;
import org.objectweb.cjdbc.controller.loadbalancer.BackendWorkerThread;
import org.objectweb.cjdbc.controller.loadbalancer.tasks.AbstractTask;
import org.objectweb.cjdbc.controller.loadbalancer.tasks.BeginTask;
import org.objectweb.cjdbc.controller.loadbalancer.tasks.KillThreadTask;

public class JDBCRecoverThread implements Runnable
{

  static Trace         logger = Trace.getLogger(JDBCRecoverThread.class
                                  .getName());

  AbstractRecoveryLog  recoveryLog;
  DatabaseBackend      backend;
  long                 logIdx;
  AbstractLoadBalancer loadBalancer;
  SQLException         exception;

  BackendWorkerThread  bwt;
  ArrayList            tids;

  public JDBCRecoverThread(AbstractRecoveryLog recoveryLog,
      DatabaseBackend backend, long recoveryTransactionId,
      AbstractLoadBalancer loadBalancer)
  {
    this.recoveryLog = recoveryLog;
    this.backend = backend;
    this.logIdx = recoveryTransactionId;
    this.loadBalancer = loadBalancer;
    tids = new ArrayList();
  }

  public long getLogIdx()
  {
    return logIdx;
  }

  public void setLogIdx(long logIdx)
  {
    this.logIdx = logIdx;
  }

  public SQLException getException()
  {
    return exception;
  }

  public void run()
  {
    try
    {
      bwt = new BackendWorkerThread("Worker thread for recovery on backend:"
          + backend.getName(), backend, loadBalancer);
      bwt.start();
      RecoveryTask recoveryTask = null;
      AbstractTask abstractTask = null;

      logger.info(Translate.get("recovery.start.process"));

      long tid;
      // Replay the whole log
      while (logIdx != -1)
      {
        try
        {
          recoveryTask = recoveryLog.recoverNextRequest(logIdx);
        }
        catch (SQLException e)
        {
          // Signal end of recovery and kill worker thread
          recoveryLog.endRecovery();
          addWorkerTask(bwt, new KillThreadTask(1, 1));
          String msg = Translate.get("recovery.cannot.recover.from.index", e);
          logger.error(msg, e);
          throw new SQLException(msg);
        }
        if (recoveryTask == null)
          break;
        tid = recoveryTask.getTid();
        if (tid == 0)
          // autocommit ok
          ;
        else
        {
          if (recoveryTask.getTask() instanceof BeginTask)
            tids.add(new Long(tid));
          else if (!tids.contains(new Long(tid)))
          {
            /*
             * if the task transaction id does not have a corresponding begin
             * (it is not in the tids arraylist), then this task has already
             * been played when the backend was disabled. So we can skip it.
             */
            logIdx++;
            continue;
          }
        }
        abstractTask = recoveryTask.getTask();
        logIdx = recoveryTask.getId();
        synchronized (abstractTask)
        {
          try
          {
            addWorkerTask(bwt, abstractTask);
            abstractTask.wait();
            if (abstractTask.getFailed() > 0)
            {
              // Signal end of recovery and kill worker thread
              recoveryLog.endRecovery();
              addWorkerTask(bwt, new KillThreadTask(1, 1));
              String msg = Translate.get("recovery.failed.with.error",
                  new String[]{
                      abstractTask.toString(),
                      ((Exception) abstractTask.getExceptions().get(0))
                          .getMessage()});
              logger.error(msg);
              throw new SQLException(msg);
            }
          }
          catch (InterruptedException e1)
          {
            // Signal end of recovery and kill worker thread
            recoveryLog.endRecovery();
            addWorkerTask(bwt, new KillThreadTask(1, 1));
            throw new SQLException(Translate.get(
                "recovery.interrupted.with.request", recoveryTask.getTask()
                    .toString()));
          }
        }
      }
    }
    catch (SQLException e)
    {
      exception = e;
    }
  }

  private void addWorkerTask(BackendWorkerThread bwt, AbstractTask task)
  {
    synchronized (bwt)
    {
      bwt.addTask(task);
      bwt.notify();
    }
  }

  public void endRecovery()
  {
    // We are done with the recovery
    logger.info(Translate.get("recovery.process.complete"));
    if (bwt != null)
    {
      addWorkerTask(bwt, new KillThreadTask(1, 1));
      try
      {
        bwt.join();
      }
      catch (InterruptedException e)
      {
        recoveryLog.endRecovery();
        String msg = Translate.get("recovery.join.failed", e);
        logger.error(msg, e);
        exception = new SQLException(msg);
      }
    }

    recoveryLog.endRecovery();
  }

}

---