Issue 2368: tasks should be interruptable. The four schedulable task (impo...

kenneth_suter

16.19.2007 eee90cc3e34f4e5a7d562ee9e60c66a61816985c

Issue 2368: tasks should be interruptable.  The four schedulable task (import-ldif, export-ldif, backup, and restore) can now be interrupted for purposes of cancellation.  The manage-tasks utility now allows the user to cancel any one of these tasks if they are currently running.  If interrupted while executing, the tasks try to break out of their work loop as soon as possible and return a 'stopped by administrator' status.  Both the backup and export-ldif tasks perform some cleanup (removing the abandoned backup or exported LDIF file) if they are cancelled.

 opends/src/messages/messages/task.properties

@@ -186,3 +186,4 @@
INFO_FAILED_DEPENDENCY_ACTION_PROCESS_99=Process
INFO_FAILED_DEPENDENCY_ACTION_CANCEL_100=Cancel
INFO_FAILED_DEPENDENCY_ACTION_DISABLE_101=Disable
INFO_TASK_STOPPED_BY_ADMIN_102=Task was stopped by an administrator:  %s

 opends/src/messages/messages/tools.properties

@@ -2188,4 +2188,4 @@
  option is to be used in conjunction with one or more dependencies
SEVERE_ERR_TASKINFO_TASK_NOT_CANCELABLE_TASK_1477=Error:  task %s is not in a \
  cancelable state

NOTICE_BACKUPDB_CANCELLED_1478=The backup process was cancelled

 opends/src/server/org/opends/server/backends/SchemaBackend.java

@@ -4277,6 +4277,11 @@
    byte[] buffer = new byte[8192];
    for (File schemaFile : schemaFiles)
    {
      if (backupConfig.isCancelled())
      {
        break;
      }

      if (! schemaFile.isFile())
      {
        // If there are any non-file items in the directory (e.g., one or more
@@ -4310,7 +4315,7 @@
        while (true)
        {
          int bytesRead = inputStream.read(buffer);
          if (bytesRead < 0)
          if (bytesRead < 0 || backupConfig.isCancelled())
          {
            break;
          }

 opends/src/server/org/opends/server/backends/jeb/BackupManager.java

@@ -476,7 +476,8 @@
      if (latestFileName != null)
      {
        ArrayList<String> unchangedList = new ArrayList<String>();
        while (indexCurrent < logFiles.length)
        while (indexCurrent < logFiles.length &&
                !backupConfig.isCancelled())
        {
          File logFile = logFiles[indexCurrent];
          String logFileName = logFile.getName();
@@ -528,13 +529,15 @@
      {
        boolean deletedFiles = false;

        while (indexCurrent < logFiles.length)
        while (indexCurrent < logFiles.length &&
                !backupConfig.isCancelled())
        {
          File logFile = logFiles[indexCurrent];

          try
          {
            latestFileSize = archiveFile(zipStream, mac, digest, logFile);
            latestFileSize = archiveFile(zipStream, mac, digest,
                                         logFile, backupConfig);
            latestFileName = logFile.getName();
          }
          catch (FileNotFoundException e)
@@ -695,6 +698,13 @@
      throw new DirectoryException(DirectoryServer.getServerErrorResultCode(),
                                   message, e);
    }

    // Remove the backup if this operation was cancelled since the
    // backup may be incomplete
    if (backupConfig.isCancelled())
    {
      removeBackup(backupDir, backupID);
    }
  }


@@ -991,7 +1001,7 @@

    // Iterate through the entries in the zip file.
    ZipEntry zipEntry = zipStream.getNextEntry();
    while (zipEntry != null)
    while (zipEntry != null && !restoreConfig.isCancelled())
    {
      String name = zipEntry.getName();

@@ -1078,7 +1088,7 @@
        long totalBytesRead = 0;
        byte[] buffer = new byte[8192];
        int bytesRead = zipStream.read(buffer);
        while (bytesRead > 0)
        while (bytesRead > 0 && !restoreConfig.isCancelled())
        {
          totalBytesRead += bytesRead;

@@ -1153,7 +1163,8 @@
   * @throws IOException If an I/O error occurs while archiving the file.
   */
  private long archiveFile(ZipOutputStream zipStream,
                           Mac mac, MessageDigest digest, File file)
                           Mac mac, MessageDigest digest, File file,
                           BackupConfig backupConfig)
       throws IOException, FileNotFoundException
  {
    ZipEntry zipEntry = new ZipEntry(file.getName());
@@ -1179,7 +1190,7 @@
    long totalBytesRead = 0;
    byte[] buffer = new byte[8192];
    int bytesRead = inputStream.read(buffer);
    while (bytesRead > 0)
    while (bytesRead > 0 && !backupConfig.isCancelled())
    {
      if (mac != null)
      {

 opends/src/server/org/opends/server/backends/jeb/ExportJob.java

@@ -143,6 +143,11 @@
    {
      for (EntryContainer exportContainer : exportContainers)
      {
        if (exportConfig.isCancelled())
        {
          break;
        }

        exportContainer.sharedLock.lock();
        try
        {
@@ -201,6 +206,11 @@
           status == OperationStatus.SUCCESS;
           status = cursor.getNext(key, data, LockMode.DEFAULT))
      {
        if (exportConfig.isCancelled())
        {
          break;
        }

        EntryID entryID = null;
        try
        {

 opends/src/server/org/opends/server/backends/jeb/ImportJob.java

@@ -601,6 +601,11 @@

    do
    {
      if (ldifImportConfig.isCancelled())
      {
        break;
      }

      if(threads.size() <= 0)
      {
        message = ERR_JEB_IMPORT_NO_WORKER_THREADS.get();
@@ -674,7 +679,8 @@
        {
          status = cursor.getFirst(key, data, lockMode);

          while(status == OperationStatus.SUCCESS)
          while(status == OperationStatus.SUCCESS &&
                !ldifImportConfig.isCancelled())
          {
            if(threads.size() <= 0)
            {
@@ -772,7 +778,8 @@
              end[0] = (byte) (end[0] + 1);

              while(status == OperationStatus.SUCCESS &&
                  dn2idComparator.compare(key.getData(), end) < 0)
                  dn2idComparator.compare(key.getData(), end) < 0 &&
                  !ldifImportConfig.isCancelled())
              {
                if(threads.size() <= 0)
                {

 opends/src/server/org/opends/server/backends/task/Task.java

@@ -142,6 +142,9 @@
  // The current state of this task.
  private TaskState taskState;

  // The task state that may be set when the task is interrupted.
  private TaskState taskInterruptState;

  // The scheduler with which this task is associated.
  private TaskScheduler taskScheduler;

@@ -577,6 +580,18 @@
    return taskState;
  }

  /**
   * Indicates whether or not this task has been cancelled.
   *
   * @return boolean where true indicates that this task was
   *         cancelled either before or during execution
   */
  public boolean isCancelled()
  {
    return taskInterruptState != null &&
      TaskState.isCancelled(taskInterruptState);
  }



  /**
@@ -626,6 +641,58 @@
    }
  }


  /**
   * Sets a state for this task that is the result of a call to
   * {@link #interruptTask(TaskState, org.opends.messages.Message)}.
   * It may take this task some time to actually cancel to that
   * actual state may differ until quiescence.
   *
   * @param state for this task once it has canceled whatever it is doing
   */
  protected void setTaskInterruptState(TaskState state)
  {
    this.taskInterruptState = state;
  }


  /**
   * Gets the interrupt state for this task that was set as a
   * result of a call to {@link #interruptTask(TaskState,
   * org.opends.messages.Message)}.
   *
   * @return interrupt state for this task
   */
  protected TaskState getTaskInterruptState()
  {
    return this.taskInterruptState;
  }


  /**
   * Returns a state for this task after processing has completed.
   * If the task was interrupted with a call to
   * {@link #interruptTask(TaskState, org.opends.messages.Message)}
   * then that method's interruptState is returned here.  Otherwse
   * this method returns TaskState.COMPLETED_SUCCESSFULLY.  It is
   * assumed that if there were errors during task processing that
   * task state will have been derived in some other way.
   *
   * @return state for this task after processing has completed
   */
  protected TaskState getFinalTaskState()
  {
    if (this.taskInterruptState == null)
    {
      return TaskState.COMPLETED_SUCCESSFULLY;
    }
    else
    {
      return this.taskInterruptState;
    }
  }


  /**
   * Replaces an attribute values of the task entry.
   *
@@ -1241,6 +1308,10 @@
   * gracefully interrupt a task, then subclasses should override this method to
   * do so.
   *
   * Implementations of this method are exprected to call
   * {@link #setTaskInterruptState(TaskState)} if the interruption is accepted
   * by this task.
   *
   * @param  interruptState   The state to use for the task if it is
   *                          successfully interrupted.
   * @param  interruptReason  A human-readable explanation for the cancellation.

 opends/src/server/org/opends/server/backends/task/TaskState.java

@@ -224,6 +224,26 @@
  }


  /**
   * Indicates whether or not this task has been cancelled.
   *
   * @param  taskState  The task state for which to make the determination.
   *
   * @return  <CODE>true</CODE> if the task state indicates that the task
   *          was cancelled either before or during execution, or
   *          <CODE>false</CODE> otherwise.
   */
  public static boolean isCancelled(TaskState taskState)
  {
    switch(taskState)
    {
      case STOPPED_BY_ADMINISTRATOR:
      case CANCELED_BEFORE_STARTING:
        return true;
      default:
        return false;
    }
  }

  /**
   * Retrieves the task state that corresponds to the provided string value.

 opends/src/server/org/opends/server/extensions/ConfigFileHandler.java

@@ -2798,7 +2798,7 @@
      while (true)
      {
        int bytesRead = inputStream.read(buffer);
        if (bytesRead < 0)
        if (bytesRead < 0 || backupConfig.isCancelled())
        {
          break;
        }
@@ -2862,7 +2862,7 @@
          while (true)
          {
            int bytesRead = inputStream.read(buffer);
            if (bytesRead < 0)
            if (bytesRead < 0 || backupConfig.isCancelled())
            {
              break;
            }
@@ -2972,6 +2972,14 @@
      throw new DirectoryException(DirectoryServer.getServerErrorResultCode(),
                                   message, e);
    }

    // Remove the backup if this operation was cancelled since the
    // backup may be incomplete
    if (backupConfig.isCancelled())
    {
      removeBackup(backupDirectory, backupID);
    }

  }



 opends/src/server/org/opends/server/replication/server/ReplicationBackend.java

@@ -505,6 +505,10 @@
    {
      for (ReplicationCache exportContainer : exportContainers)
      {
        if (exportConfig.isCancelled())
        {
          break;
        }
        processContainer(exportContainer, exportConfig, ldifWriter, null);
      }
    }
@@ -571,6 +575,11 @@

    for (ReplicationCache exportContainer : exportContainers)
    {
      if (exportConfig != null && exportConfig.isCancelled())
      {
        break;
      }

      attributes.clear();
      ldapAttrList.clear();

@@ -630,6 +639,11 @@
    // Walk through the servers
    for (Short serverId : rc.getServers())
    {
      if (exportConfig != null && exportConfig.isCancelled())
      {
        break;
      }

      ReplicationIterator ri = rc.getChangelogIterator(serverId,
          null);

@@ -640,6 +654,10 @@
          // Walk through the changes
          while (ri.getChange() != null)
          {
            if (exportConfig != null && exportConfig.isCancelled())
            {
              break;
            }
            UpdateMessage msg = ri.getChange();
            processChange(msg, exportConfig, ldifWriter, searchOperation);
            if (!ri.next())

 opends/src/server/org/opends/server/tasks/BackupTask.java

@@ -26,6 +26,7 @@
 */
package org.opends.server.tasks;
import org.opends.messages.Message;
import org.opends.messages.TaskMessages;

import static org.opends.server.config.ConfigConstants.*;
import static org.opends.server.core.DirectoryServer.getAttributeType;
@@ -133,6 +134,8 @@
  private File    backupDirectory;
  private String  incrementalBase;

  private BackupConfig backupConfig;

  /**
   * All the backend configuration entries defined in the server mapped
   * by their backend ID.
@@ -460,7 +463,7 @@


    // Create a backup configuration.
    BackupConfig backupConfig = new BackupConfig(backupDir, backupID,
    backupConfig = new BackupConfig(backupDir, backupID,
                                                 incremental);
    backupConfig.setCompressData(compress);
    backupConfig.setEncryptData(encrypt);
@@ -560,6 +563,30 @@
  /**
   * {@inheritDoc}
   */
  public void interruptTask(TaskState interruptState, Message interruptReason)
  {
    if (TaskState.STOPPED_BY_ADMINISTRATOR.equals(interruptState) &&
            backupConfig != null)
    {
      addLogMessage(TaskMessages.INFO_TASK_STOPPED_BY_ADMIN.get(
              interruptReason));
      setTaskInterruptState(interruptState);
      backupConfig.cancel();
    }
  }


  /**
   * {@inheritDoc}
   */
  public boolean isInterruptable() {
    return true;
  }


  /**
   * {@inheritDoc}
   */
  protected TaskState runTask()
  {
    if (!argumentsAreValid())
@@ -585,6 +612,11 @@
    boolean errorsEncountered = false;
    for (Backend b : backendsToArchive)
    {
      if (isCancelled())
      {
        break;
      }

      // Acquire a shared lock for this backend.
      if (!lockBackend(b))
      {
@@ -639,6 +671,12 @@
      logError(message);
      return TaskState.COMPLETED_WITH_ERRORS;
    }
    else if (isCancelled())
    {
      Message message = NOTE_BACKUPDB_CANCELLED.get();
      logError(message);
      return getTaskInterruptState();
    }
    else
    {
      Message message = NOTE_BACKUPDB_COMPLETED_SUCCESSFULLY.get();

 opends/src/server/org/opends/server/tasks/ExportTask.java

@@ -26,6 +26,7 @@
 */
package org.opends.server.tasks;
import org.opends.messages.Message;
import org.opends.messages.TaskMessages;

import static org.opends.server.core.DirectoryServer.getAttributeType;
import static org.opends.server.config.ConfigConstants.*;
@@ -55,6 +56,7 @@
import java.util.List;
import java.util.Map;
import java.util.HashMap;
import java.io.File;

/**
 * This class provides an implementation of a Directory Server task that can
@@ -136,6 +138,7 @@
  private ArrayList<String> includeBranchStrings;
  private ArrayList<String> excludeBranchStrings;

  private LDIFExportConfig exportConfig;

  /**
   * {@inheritDoc}
@@ -259,6 +262,31 @@

  }


  /**
   * {@inheritDoc}
   */
  public void interruptTask(TaskState interruptState, Message interruptReason)
  {
    if (TaskState.STOPPED_BY_ADMINISTRATOR.equals(interruptState) &&
            exportConfig != null)
    {
      addLogMessage(TaskMessages.INFO_TASK_STOPPED_BY_ADMIN.get(
              interruptReason));
      setTaskInterruptState(interruptState);
      exportConfig.cancel();
    }
  }


  /**
   * {@inheritDoc}
   */
  public boolean isInterruptable() {
    return true;
  }


  /**
   * {@inheritDoc}
   */
@@ -483,8 +511,7 @@
      existingBehavior = ExistingFileBehavior.OVERWRITE;
    }

    LDIFExportConfig exportConfig =
         new LDIFExportConfig(ldifFile, existingBehavior);
    exportConfig = new LDIFExportConfig(ldifFile, existingBehavior);
    exportConfig.setCompressData(compressLDIF);
    exportConfig.setEncryptData(encryptLDIF);
    exportConfig.setExcludeAttributes(excludeAttributes);
@@ -587,7 +614,19 @@
      exportConfig.close();
    }

    // If the operation was cancelled delete the export file since
    // if will be incomplete.
    if (exportConfig.isCancelled())
    {
      File f = new File(ldifFile);
      if (f.exists())
      {
        f.delete();
      }
    }

    return TaskState.COMPLETED_SUCCESSFULLY;
    // If we got here the task either completed successfully or
    // was interrupted
    return getFinalTaskState();
  }
}

 opends/src/server/org/opends/server/tasks/ImportTask.java

@@ -26,6 +26,7 @@
 */
package org.opends.server.tasks;
import org.opends.messages.Message;
import org.opends.messages.TaskMessages;

import static org.opends.messages.TaskMessages.*;
import static org.opends.messages.ToolMessages.*;
@@ -169,6 +170,7 @@
  ArrayList<String>  includeFilterStrings    = null;
  ArrayList<String>  ldifFiles               = null;

  private LDIFImportConfig importConfig;

  /**
   * {@inheritDoc}
@@ -482,6 +484,30 @@
  }


  /**
   * {@inheritDoc}
   */
  public void interruptTask(TaskState interruptState, Message interruptReason)
  {
    if (TaskState.STOPPED_BY_ADMINISTRATOR.equals(interruptState) &&
            importConfig != null)
    {
      addLogMessage(TaskMessages.INFO_TASK_STOPPED_BY_ADMIN.get(
              interruptReason));
      setTaskInterruptState(interruptState);
      importConfig.cancel();
    }
  }


  /**
   * {@inheritDoc}
   */
  public boolean isInterruptable()
  {
    return true;
  }


  /**
   * {@inheritDoc}
@@ -731,7 +757,7 @@

    // Create the LDIF import configuration to use when reading the LDIF.
    ArrayList<String> fileList = new ArrayList<String>(ldifFiles);
    LDIFImportConfig importConfig = new LDIFImportConfig(fileList);
    importConfig = new LDIFImportConfig(fileList);
    importConfig.setAppendToExistingData(append);
    importConfig.setReplaceExistingEntries(replaceExisting);
    importConfig.setCompressed(isCompressed);
@@ -942,6 +968,6 @@

    // Clean up after the import by closing the import config.
    importConfig.close();
    return TaskState.COMPLETED_SUCCESSFULLY;
    return getFinalTaskState();
  }
}

 opends/src/server/org/opends/server/tasks/RestoreTask.java

@@ -26,6 +26,7 @@
 */
package org.opends.server.tasks;
import org.opends.messages.Message;
import org.opends.messages.TaskMessages;

import static org.opends.server.core.DirectoryServer.getAttributeType;
import static org.opends.server.config.ConfigConstants.*;
@@ -100,6 +101,7 @@
  private String backupID;
  private boolean verifyOnly;

  private RestoreConfig restoreConfig;

  /**
   * {@inheritDoc}
@@ -223,6 +225,31 @@
    return true;
  }


  /**
   * {@inheritDoc}
   */
  public void interruptTask(TaskState interruptState, Message interruptReason)
  {
    if (TaskState.STOPPED_BY_ADMINISTRATOR.equals(interruptState) &&
            restoreConfig != null)
    {
      addLogMessage(TaskMessages.INFO_TASK_STOPPED_BY_ADMIN.get(
              interruptReason));
      setTaskInterruptState(interruptState);
      restoreConfig.cancel();
    }
  }


  /**
   * {@inheritDoc}
   */
  public boolean isInterruptable() {
    return true;
  }


  /**
   * {@inheritDoc}
   */
@@ -311,8 +338,7 @@
    }

    // Create the restore config object from the information available.
    RestoreConfig restoreConfig = new RestoreConfig(backupDir, backupID,
                                                    verifyOnly);
    restoreConfig = new RestoreConfig(backupDir, backupID, verifyOnly);

    // Notify the task listeners that a restore is going to start
    // this must be done before disabling the backend to allow
@@ -408,7 +434,7 @@
    }
    else
    {
      return TaskState.COMPLETED_SUCCESSFULLY;
      return getFinalTaskState();
    }
  }
}

 opends/src/server/org/opends/server/types/BackupConfig.java

@@ -49,7 +49,7 @@
     mayInstantiate=true,
     mayExtend=false,
     mayInvoke=true)
public final class BackupConfig
public final class BackupConfig extends OperationConfig
{
  // The path to the directory in which the backup file(s) should be
  // created.

 opends/src/server/org/opends/server/types/LDIFExportConfig.java

@@ -56,7 +56,7 @@
     mayInstantiate=true,
     mayExtend=false,
     mayInvoke=true)
public final class LDIFExportConfig
public final class LDIFExportConfig extends OperationConfig
{
  /**
   * The tracer object for the debug logger.

 opends/src/server/org/opends/server/types/LDIFImportConfig.java

@@ -64,7 +64,7 @@
     mayInstantiate=true,
     mayExtend=false,
     mayInvoke=true)
public final class LDIFImportConfig
public final class LDIFImportConfig extends OperationConfig
{
  /**
   * The tracer object for the debug logger.

 opends/src/server/org/opends/server/types/OperationConfig.java

New file
@@ -0,0 +1,66 @@
/*
 * CDDL HEADER START
 *
 * The contents of this file are subject to the terms of the
 * Common Development and Distribution License, Version 1.0 only
 * (the "License").  You may not use this file except in compliance
 * with the License.
 *
 * You can obtain a copy of the license at
 * trunk/opends/resource/legal-notices/OpenDS.LICENSE
 * or https://OpenDS.dev.java.net/OpenDS.LICENSE.
 * See the License for the specific language governing permissions
 * and limitations under the License.
 *
 * When distributing Covered Code, include this CDDL HEADER in each
 * file and include the License file at
 * trunk/opends/resource/legal-notices/OpenDS.LICENSE.  If applicable,
 * add the following below this CDDL HEADER, with the fields enclosed
 * by brackets "[]" replaced with your own identifying information:
 *      Portions Copyright [yyyy] [name of copyright owner]
 *
 * CDDL HEADER END
 *
 *
 *      Portions Copyright 2007 Sun Microsystems, Inc.
 */

package org.opends.server.types;

/**
 * Base for data structures that define configuration
 * for operations.
 */
@org.opends.server.types.PublicAPI(
     stability=org.opends.server.types.StabilityLevel.VOLATILE,
     mayInstantiate=true,
     mayExtend=false,
     mayInvoke=true)
public abstract class OperationConfig {

  // When true indicates that the operation should stop as soon as
  // possible.
  private boolean cancelled;

  /**
   * Indicates that this operation has been cancelled and the
   * operation if executing should finish as soon as possible.
   */
  public void cancel()
  {
    this.cancelled = true;
  }

  /**
   * Indicates whether or not this operation has been
   * cancelled.
   *
   * @return boolean where true indicates that this
   *         operation has been cancelled and if currently
   *         executing will finish as soon as possible
   */
  public boolean isCancelled()
  {
    return this.cancelled;
  }
}

 opends/src/server/org/opends/server/types/RestoreConfig.java

@@ -47,7 +47,7 @@
     mayInstantiate=true,
     mayExtend=false,
     mayInvoke=true)
public final class RestoreConfig
public final class RestoreConfig extends OperationConfig
{
  // The reference to the directory containing the backup file(s) to
  // restore.