OPENDJ-1231 Make the Medium Consistency Point support replica heartbeats

Jean-Noel Rouvignac

17.05.2013 d356a45009acfa47a94f3d023ac38b4c1faadd13

 opends/src/server/org/opends/server/replication/server/changelog/je/ReplicationDB.java

@@ -59,8 +59,6 @@
 */
public class ReplicationDB
{
  private static final int START = 0;
  private static final int STOP = 1;

  private Database db;
  private ReplicationDbEnv dbenv;
@@ -901,236 +899,6 @@
  }

  /**
   * Count the number of changes between 2 changes numbers (inclusive).
   * @param start The lower limit of the count.
   * @param stop The higher limit of the count.
   * @return The number of changes between provided start and stop CSN.
   * Returns 0 when an error occurs.
   */
  public long count(CSN start, CSN stop)
  {
    dbCloseLock.readLock().lock();
    try
    {
      // If the DB has been closed then return immediately.
      if (isDBClosed())
      {
        return 0;
      }
      if (start == null && stop == null)
      {
        return db.count();
      }

      int[] counterValues = new int[2];
      int[] distanceToCounterRecords = new int[2];

      // Step 1 : from the start point, traverse db to the next counter record
      // or to the stop point.
      findFirstCounterRecordAfterStartPoint(start, stop, counterValues,
          distanceToCounterRecords);

      // cases
      if (counterValues[START] == 0)
        return distanceToCounterRecords[START];

      // Step 2 : from the stop point, traverse db to the next counter record
      // or to the start point.
      if (!findFirstCounterRecordBeforeStopPoint(start, stop, counterValues,
          distanceToCounterRecords))
      {
        // database is empty
        return 0;
      }

      // Step 3 : Now consolidates the result
      return computeDistance(counterValues, distanceToCounterRecords);
    }
    catch (DatabaseException e)
    {
      dbenv.shutdownOnException(e);
    }
    finally
    {
      dbCloseLock.readLock().unlock();
    }
    return 0;
  }

  private void findFirstCounterRecordAfterStartPoint(CSN start, CSN stop,
      int[] counterValues, int[] distanceToCounterRecords)
      throws DatabaseException
  {
    Cursor cursor = db.openCursor(null, null);
    try
    {
      OperationStatus status;
      DatabaseEntry key;
      DatabaseEntry data = new DatabaseEntry();
      if (start != null)
      {
        key = createReplicationKey(start);
        status = cursor.getSearchKey(key, data, LockMode.DEFAULT);
        if (status == OperationStatus.NOTFOUND)
          status = cursor.getSearchKeyRange(key, data, LockMode.DEFAULT);
      }
      else
      {
        key = new DatabaseEntry();
        // JNR: I suspect this is equivalent to writing cursor.getFirst().
        // If it is, then please change the code to make it clearer.
        status = cursor.getNext(key, data, LockMode.DEFAULT);
      }

      while (status == OperationStatus.SUCCESS)
      {
        // test whether the record is a regular change or a counter
        final CSN csn = toCSN(key.getData());
        if (isACounterRecord(csn))
        {
          // we have found the counter record
          counterValues[START] = decodeCounterValue(data.getData());
          break;
        }

        // it is a regular change record
        if (csn.isNewerThan(stop))
        { // we are outside the range: we reached the 'stop' target
          break;
        }

        distanceToCounterRecords[START]++;
        status = cursor.getNext(key, data, LockMode.DEFAULT);
        // loop to update the distance and possibly find a counter record
      }
    }
    finally
    {
      close(cursor);
    }
  }

  private boolean findFirstCounterRecordBeforeStopPoint(CSN start, CSN stop,
      int[] counterValues, int[] distanceToCounterRecords)
      throws DatabaseException
  {
    Cursor cursor = db.openCursor(null, null);
    try
    {
      DatabaseEntry key = createReplicationKey(stop);
      DatabaseEntry data = new DatabaseEntry();
      OperationStatus status = cursor.getSearchKey(key, data, LockMode.DEFAULT);
      if (status != OperationStatus.SUCCESS)
      {
        key = new DatabaseEntry();
        data = new DatabaseEntry();
        status = cursor.getLast(key, data, LockMode.DEFAULT);
        if (status != OperationStatus.SUCCESS)
        {
          return false;
        }
      }

      while (status == OperationStatus.SUCCESS)
      {
        final CSN csn = toCSN(key.getData());
        if (isACounterRecord(csn))
        {
          // we have found the counter record
          counterValues[STOP] = decodeCounterValue(data.getData());
          break;
        }

        // it is a regular change record
        if (csn.isOlderThan(start))
        { // we are outside the range: we reached the 'start' target
          break;
        }

        distanceToCounterRecords[STOP]++;
        status = cursor.getPrev(key, data, LockMode.DEFAULT);
        // loop to update the distance and possibly find a counter record
      }
      return true;
    }
    finally
    {
      close(cursor);
    }
  }

  /**
   * The diagram below shows a visual description of how the distance between
   * two CSNs in the database is computed.
   *
   * <pre>
   *     +--------+                        +--------+
   *     | CASE 1 |                        | CASE 2 |
   *     +--------+                        +--------+
   *
   *             CSN                               CSN
   *             -----                             -----
   *   START  => -----                   START  => -----
   *     ^       -----                     ^       -----
   *     |       -----                     |       -----
   *   dist 1    -----                   dist 1    -----
   *     |       -----                     |       -----
   *     v       -----                     v       -----
   *   CR 1&2 => [1000]                   CR 1  => [1000]
   *     ^       -----                             -----
   *     |       -----                             -----
   *   dist 2    -----                             -----
   *     |       -----                             -----
   *     v       -----                             -----
   *   STOP   => -----                             -----
   *             -----                             -----
   *     CR   => [2000]                   CR 2  => [2000]
   *             -----                     ^       -----
   *                                       |       -----
   *                                     dist 2    -----
   *                                       |       -----
   *                                       v       -----
   *                                     STOP   => -----
   * </pre>
   *
   * Explanation of the terms used:
   * <dl>
   * <dt>START</dt>
   * <dd>Start CSN for the count</dd>
   * <dt>STOP</dt>
   * <dd>Stop CSN for the count</dd>
   * <dt>dist</dt>
   * <dd>Distance from START (or STOP) to the counter record</dd>
   * <dt>CSN</dt>
   * <dd>Stands for "Change Sequence Number". Below it, the database is
   * symbolized, where each record is represented by using dashes "-----". The
   * database is ordered.</dd>
   * <dt>CR</dt>
   * <dd>Stands for "Counter Record". Counter Records are inserted in the
   * database along with real CSNs, but they are not real changes. They are only
   * used to speed up calculating the distance between 2 CSNs without the need
   * to scan the whole database in between.</dd>
   * </dl>
   */
  private long computeDistance(int[] counterValues,
      int[] distanceToCounterRecords)
  {
    if (counterValues[START] != 0)
    {
      if (counterValues[START] == counterValues[STOP])
      {
        // only one counter record between from and to - no need to use it
        return distanceToCounterRecords[START] + distanceToCounterRecords[STOP];
      }
      // at least 2 counter records between from and to
      return distanceToCounterRecords[START]
          + (counterValues[STOP] - counterValues[START])
          + distanceToCounterRecords[STOP];
    }
    return 0;
  }

  /**
   * Whether a provided CSN represents a counter record. A counter record is
   * used to store the time.
   *