Update the task backend to properly acquire read locks on task and recurrin...

neil_a_wilson

03.12.2007 4698e4fa18f652019b8f06fef1b2b8813446f5e0

Update the task backend to properly acquire read locks on task and recurring
task entries when evaluating them during search operations, and to acquire
write locks for the entries when they are being updated.  This should eliminate
the possibility of concurrent modification exceptions being thrown in cases
where a client issues a search for a task entry at the same time the task entry
is being updated within the server.

OpenDS Issue Number:  858

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

@@ -37,6 +37,7 @@
import org.opends.server.types.AttributeType;
import org.opends.server.types.AttributeValue;
import org.opends.server.types.DirectoryException;
import org.opends.server.types.DN;
import org.opends.server.types.Entry;
import org.opends.server.types.InitializationException;
import org.opends.server.types.ResultCode;
@@ -63,6 +64,9 @@



  // The DN of the entry that actually defines this task.
  private DN recurringTaskEntryDN;

  // The entry that actually defines this task.
  private Entry recurringTaskEntry;

@@ -99,6 +103,7 @@

    this.taskScheduler      = taskScheduler;
    this.recurringTaskEntry = recurringTaskEntry;
    this.recurringTaskEntryDN = recurringTaskEntry.getDN();


    // Get the recurring task ID from the entry.  If there isn't one, then fail.
@@ -278,6 +283,20 @@


  /**
   * Retrieves the DN of the entry containing the data for this recurring task.
   *
   * @return  The DN of the entry containing the data for this recurring task.
   */
  public DN getRecurringTaskEntryDN()
  {
    assert debugEnter(CLASS_NAME, "getRecurringTaskEntryDN");

    return recurringTaskEntryDN;
  }



  /**
   * Retrieves the entry containing the data for this recurring task.
   *
   * @return  The entry containing the data for this recurring task.

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

@@ -37,6 +37,7 @@
import java.util.List;
import java.util.TimeZone;
import java.util.UUID;
import java.util.concurrent.locks.Lock;

import org.opends.server.core.DirectoryServer;
import org.opends.server.protocols.asn1.ASN1OctetString;
@@ -44,6 +45,7 @@
import org.opends.server.types.AttributeType;
import org.opends.server.types.AttributeValue;
import org.opends.server.types.DirectoryException;
import org.opends.server.types.DN;
import org.opends.server.types.Entry;
import org.opends.server.types.ErrorLogCategory;
import org.opends.server.types.ErrorLogSeverity;
@@ -75,6 +77,9 @@



  // The DN for the task entry.
  private DN taskEntryDN;

  // The entry that actually defines this task.
  private Entry taskEntry;

@@ -141,8 +146,9 @@

    this.taskScheduler = taskScheduler;
    this.taskEntry     = taskEntry;
    this.taskEntryDN   = taskEntry.getDN();

    String taskDN = taskEntry.getDN().toString();
    String taskDN = taskEntryDN.toString();

    logMessageCounter = 0;

@@ -438,6 +444,20 @@


  /**
   * Retrieves the DN of the entry containing the definition for this task.
   *
   * @return  The DN of the entry containing the definition for this task.
   */
  public final DN getTaskEntryDN()
  {
    assert debugEnter(CLASS_NAME, "getTaskEntryDN");

    return taskEntryDN;
  }



  /**
   * Retrieves the entry containing the definition for this task.
   *
   * @return  The entry containing the definition for this task.
@@ -507,6 +527,10 @@
  {
    assert debugEnter(CLASS_NAME, "setTaskState", String.valueOf(taskState));

    Lock lock = taskScheduler.writeLockEntry(taskEntryDN);

    try
    {
    this.taskState = taskState;

    AttributeType type =
@@ -516,7 +540,8 @@
      type = DirectoryServer.getDefaultAttributeType(ATTR_TASK_STATE);
    }

    LinkedHashSet<AttributeValue> values = new LinkedHashSet<AttributeValue>();
      LinkedHashSet<AttributeValue> values =
           new LinkedHashSet<AttributeValue>();
    values.add(new AttributeValue(type,
                                  new ASN1OctetString(taskState.toString())));

@@ -524,6 +549,11 @@
    attrList.add(new Attribute(type, ATTR_TASK_STATE, values));
    taskEntry.putAttribute(type, attrList);
  }
    finally
    {
      taskScheduler.unlockEntry(taskEntryDN, lock);
    }
  }



@@ -574,6 +604,10 @@
    assert debugEnter(CLASS_NAME, "setActualStartTime",
                      String.valueOf(actualStartTime));

    Lock lock = taskScheduler.writeLockEntry(taskEntryDN);

    try
    {
    this.actualStartTime = actualStartTime;

    AttributeType type = DirectoryServer.getAttributeType(
@@ -589,13 +623,19 @@
    Date d = new Date(actualStartTime);
    ASN1OctetString s = new ASN1OctetString(dateFormat.format(d));

    LinkedHashSet<AttributeValue> values = new LinkedHashSet<AttributeValue>();
      LinkedHashSet<AttributeValue> values =
           new LinkedHashSet<AttributeValue>();
    values.add(new AttributeValue(type, s));

    ArrayList<Attribute> attrList = new ArrayList<Attribute>(1);
    attrList.add(new Attribute(type, ATTR_TASK_ACTUAL_START_TIME, values));
    taskEntry.putAttribute(type, attrList);
  }
    finally
    {
      taskScheduler.unlockEntry(taskEntryDN, lock);
    }
  }



@@ -629,13 +669,18 @@
    assert debugEnter(CLASS_NAME, "setCompletionTime",
                      String.valueOf(completionTime));

    Lock lock = taskScheduler.writeLockEntry(taskEntryDN);

    try
    {
    this.completionTime = completionTime;

    AttributeType type = DirectoryServer.getAttributeType(
                              ATTR_TASK_COMPLETION_TIME.toLowerCase());
    if (type == null)
    {
      type = DirectoryServer.getDefaultAttributeType(ATTR_TASK_COMPLETION_TIME);
        type =
             DirectoryServer.getDefaultAttributeType(ATTR_TASK_COMPLETION_TIME);
    }

    SimpleDateFormat dateFormat = new SimpleDateFormat(DATE_FORMAT_UTC_TIME);
@@ -643,13 +688,19 @@
    Date d = new Date(completionTime);
    ASN1OctetString s = new ASN1OctetString(dateFormat.format(d));

    LinkedHashSet<AttributeValue> values = new LinkedHashSet<AttributeValue>();
      LinkedHashSet<AttributeValue> values =
           new LinkedHashSet<AttributeValue>();
    values.add(new AttributeValue(type, s));

    ArrayList<Attribute> attrList = new ArrayList<Attribute>(1);
    attrList.add(new Attribute(type, ATTR_TASK_COMPLETION_TIME, values));
    taskEntry.putAttribute(type, attrList);
  }
    finally
    {
      taskScheduler.unlockEntry(taskEntryDN, lock);
    }
  }



@@ -756,10 +807,11 @@
                      String.valueOf(severity), String.valueOf(messageID),
                      String.valueOf(messageString));

    StringBuilder buffer = new StringBuilder();
    Lock lock = taskScheduler.writeLockEntry(taskEntryDN);

    synchronized (logMessages)
    try
    {
      StringBuilder buffer = new StringBuilder();
      buffer.append("[");
      buffer.append(TimeThread.getLocalTime());
      buffer.append("] severity=\"");
@@ -809,6 +861,10 @@
        attrList.add(new Attribute(type, ATTR_TASK_LOG_MESSAGES, values));
      }
    }
    finally
    {
      taskScheduler.unlockEntry(taskEntryDN, lock);
    }
  }



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

@@ -34,6 +34,7 @@
import java.util.LinkedHashMap;
import java.util.LinkedList;
import java.util.List;
import java.util.concurrent.locks.Lock;

import org.opends.server.api.Backend;
import org.opends.server.api.ConfigurableComponent;
@@ -888,6 +889,10 @@
      }
      else if (parentDN.equals(scheduledTaskParentDN))
      {
        Lock lock = taskScheduler.readLockEntry(baseDN);

        try
        {
        Entry e = taskScheduler.getScheduledTaskEntry(baseDN);
        if (e == null)
        {
@@ -906,8 +911,17 @@

        return;
      }
        finally
        {
          taskScheduler.unlockEntry(baseDN, lock);
        }
      }
      else if (parentDN.equals(recurringTaskParentDN))
      {
        Lock lock = taskScheduler.readLockEntry(baseDN);

        try
        {
        Entry e = taskScheduler.getRecurringTaskEntry(baseDN);
        if (e == null)
        {
@@ -926,6 +940,11 @@

        return;
      }
        finally
        {
          taskScheduler.unlockEntry(baseDN, lock);
        }
      }
      else
      {
        int    msgID   = MSGID_TASKBE_SEARCH_INVALID_BASE;

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

@@ -37,6 +37,7 @@
import java.util.LinkedList;
import java.util.List;
import java.util.TreeSet;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;

import org.opends.server.api.AlertGenerator;
@@ -55,6 +56,7 @@
import org.opends.server.types.InitializationException;
import org.opends.server.types.LDIFImportConfig;
import org.opends.server.types.LDIFExportConfig;
import org.opends.server.types.LockManager;
import org.opends.server.types.ResultCode;
import org.opends.server.types.SearchFilter;
import org.opends.server.util.LDIFException;
@@ -1448,7 +1450,83 @@


  /**
   * Retrieves the scheduled task entry with the provided DN.
   * Attempts to acquire a write lock on the specified entry, trying as many
   * times as necessary until the lock has been acquired.
   *
   * @param  entryDN  The DN of the entry for which to acquire the write lock.
   *
   * @return  The write lock that has been acquired for the entry.
   */
  Lock writeLockEntry(DN entryDN)
  {
    assert debugEnter(CLASS_NAME, "lockEntry", String.valueOf(entryDN));

    Lock lock = LockManager.lockWrite(entryDN);
    while (lock == null)
    {
      lock = LockManager.lockWrite(entryDN);
    }

    return lock;
  }



  /**
   * Attempts to acquire a read lock on the specified entry, trying up to five
   * times before failing.
   *
   * @param  entryDN  The DN of the entry for which to acquire the read lock.
   *
   * @return  The read lock that has been acquired for the entry.
   *
   * @throws  DirectoryException  If the read lock cannot be acquired.
   */
  Lock readLockEntry(DN entryDN)
       throws DirectoryException
  {
    assert debugEnter(CLASS_NAME, "lockEntry", String.valueOf(entryDN));

    Lock lock = LockManager.lockRead(entryDN);
    for (int i=0; ((lock == null) && (i < 4)); i++)
    {
      lock = LockManager.lockRead(entryDN);
    }

    if (lock == null)
    {
      int    msgID   = MSGID_BACKEND_CANNOT_LOCK_ENTRY;
      String message = getMessage(msgID, String.valueOf(entryDN));
      throw new DirectoryException(DirectoryServer.getServerErrorResultCode(),
                                   message, msgID);
    }
    else
    {
      return lock;
    }
  }



  /**
   * Releases the lock held on the specified entry.
   *
   * @param  entryDN  The DN of the entry for which the lock is held.
   * @param  lock     The lock held on the entry.
   */
  void unlockEntry(DN entryDN, Lock lock)
  {
    assert debugEnter(CLASS_NAME, "unlockEntry", String.valueOf(entryDN),
                      String.valueOf(lock));

    LockManager.unlock(entryDN, lock);
  }



  /**
   * Retrieves the scheduled task entry with the provided DN.  The caller should
   * hold a read lock on the target entry.
   *
   * @param  scheduledTaskEntryDN  The entry DN that indicates which scheduled
   *                               task entry to retrieve.
@@ -1516,6 +1594,11 @@
    {
      for (Task t : tasks.values())
      {
        DN taskEntryDN = t.getTaskEntryDN();
        Lock lock = readLockEntry(taskEntryDN);

        try
        {
        Entry e = t.getTaskEntry();
        if (filter.matchesEntry(e))
        {
@@ -1525,6 +1608,11 @@
          }
        }
      }
        finally
        {
          unlockEntry(taskEntryDN, lock);
        }
      }

      return true;
    }
@@ -1601,7 +1689,8 @@


  /**
   * Retrieves the recurring task entry with the provided DN.
   * Retrieves the recurring task entry with the provided DN.  The caller should
   * hold a read lock on the target entry.
   *
   * @param  recurringTaskEntryDN  The entry DN that indicates which recurring
   *                               task entry to retrieve.
@@ -1669,6 +1758,11 @@
    {
      for (RecurringTask rt : recurringTasks.values())
      {
        DN recurringTaskEntryDN = rt.getRecurringTaskEntryDN();
        Lock lock = readLockEntry(recurringTaskEntryDN);

        try
        {
        Entry e = rt.getRecurringTaskEntry();
        if (filter.matchesEntry(e))
        {
@@ -1678,6 +1772,11 @@
          }
        }
      }
        finally
        {
          unlockEntry(recurringTaskEntryDN, lock);
        }
      }

      return true;
    }