Replication Cleanup.

Ludovic Poitou

24.44.2013 46fd9423ab622d7f9531aa1564846ec52fe09534

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

@@ -23,7 +23,7 @@
 *
 *
 *      Copyright 2009-2010 Sun Microsystems, Inc.
 *      Portions Copyright 2012 ForgeRock AS
 *      Portions Copyright 2012-2013 ForgeRock AS
 */
package org.opends.server.replication.server;

@@ -84,11 +84,6 @@
  private ConcurrentHashMap<Integer, Long> missingChanges =
    new ConcurrentHashMap<Integer, Long>();

  // For each RS server, an approximation of the date of the first missing
  // change
  private ConcurrentHashMap<Integer, Long> fmRSDate =
    new ConcurrentHashMap<Integer, Long>();

  private ConcurrentHashMap<Integer, Long> missingChangesRS =
    new ConcurrentHashMap<Integer, Long>();

@@ -162,47 +157,38 @@
    //   Regarding each other LSj
    //    Sum the difference : max(LSj) - state(LSi)

    Iterator<Integer> lsiStateItr = this.LDAPStates.keySet().iterator();
    while (lsiStateItr.hasNext())
    {
      Integer lsiSid = lsiStateItr.next();
    for (Integer lsiSid : this.LDAPStates.keySet()) {
      ServerState lsiState = this.LDAPStates.get(lsiSid);
      Long lsiMissingChanges = (long)0;
      if (lsiState != null)
      {
        Iterator<Integer> lsjMaxItr = this.maxCNs.keySet().iterator();
        while (lsjMaxItr.hasNext())
        {
          Integer lsjSid = lsjMaxItr.next();
      Long lsiMissingChanges = (long) 0;
      if (lsiState != null) {
        for (Integer lsjSid : this.maxCNs.keySet()) {
          ChangeNumber lsjMaxCN = this.maxCNs.get(lsjSid);
          ChangeNumber lsiLastCN = lsiState.getMaxChangeNumber(lsjSid);

          int missingChangesLsiLsj =
            ChangeNumber.diffSeqNum(lsjMaxCN, lsiLastCN);
              ChangeNumber.diffSeqNum(lsjMaxCN, lsiLastCN);

          if (debugEnabled())
          {
          if (debugEnabled()) {
            mds +=
              "+ diff("+lsjMaxCN+"-"
                       +lsiLastCN+")="+missingChangesLsiLsj;
                "+ diff(" + lsjMaxCN + "-"
                    + lsiLastCN + ")=" + missingChangesLsiLsj;
          }

          // Regarding a DS that is generating changes. If it is a local DS1,
          // we get its server state, store it, then retrieve server states of
          // remote DSs. When a remote server state is coming, it may contain
          // a change number for DS1 which is newer than the one we locally
          // stored in the server state of DS1. To prevent seeing DS1 has
          // missing changes whereas it is wrong, we replace the value with 0
          // if it is a low value. We cannot overwrite big values as they may be
          // useful for a local server retrieving changes it generated earlier,
          // when it is recovering from an old snapshot and the local RS is
          // sending him the changes it is missing.
          /*
          Regarding a DS that is generating changes. If it is a local DS1,
          we get its server state, store it, then retrieve server states of
          remote DSs. When a remote server state is coming, it may contain
          a change number for DS1 which is newer than the one we locally
          stored in the server state of DS1. To prevent seeing DS1 has
          missing changes whereas it is wrong, we replace the value with 0
          if it is a low value. We cannot overwrite big values as they may be
          useful for a local server retrieving changes it generated earlier,
          when it is recovering from an old snapshot and the local RS is
          sending him the changes it is missing.
          */
          if (lsjSid.equals(lsiSid)) {
            if (missingChangesLsiLsj <= 50)
            {
            if (missingChangesLsiLsj <= 50) {
              missingChangesLsiLsj = 0;
              if (debugEnabled())
              {
              if (debugEnabled()) {
                mds += " (diff replaced by 0 as for server id " + lsiSid + ")";
              }
            }
@@ -211,11 +197,10 @@
          lsiMissingChanges += missingChangesLsiLsj;
        }
      }
      if (debugEnabled())
      {
      if (debugEnabled()) {
        mds += "=" + lsiMissingChanges;
      }
      this.missingChanges.put(lsiSid,lsiMissingChanges);
      this.missingChanges.put(lsiSid, lsiMissingChanges);
    }

    // Computes the missing changes counters for RS :
@@ -227,21 +212,17 @@
      Long lsiMissingChanges = (long)0;
      if (lsiState != null)
      {
        Iterator<Integer> lsjMaxItr = this.maxCNs.keySet().iterator();
        while (lsjMaxItr.hasNext())
        {
          int lsjSid = lsjMaxItr.next();
        for (Integer lsjSid : this.maxCNs.keySet()) {
          ChangeNumber lsjMaxCN = this.maxCNs.get(lsjSid);
          ChangeNumber lsiLastCN = lsiState.getMaxChangeNumber(lsjSid);

          int missingChangesLsiLsj =
            ChangeNumber.diffSeqNum(lsjMaxCN, lsiLastCN);
              ChangeNumber.diffSeqNum(lsjMaxCN, lsiLastCN);

          if (debugEnabled())
          {
          if (debugEnabled()) {
            mds +=
              "+ diff("+lsjMaxCN+"-"
                       +lsiLastCN+")="+missingChangesLsiLsj;
                "+ diff(" + lsjMaxCN + "-"
                    + lsiLastCN + ")=" + missingChangesLsiLsj;
          }
          lsiMissingChanges += missingChangesLsiLsj;
        }
@@ -269,42 +250,25 @@
  {
    String mds = "Monitor data=\n";

    // RS data
    Iterator<Integer> rsite = fmRSDate.keySet().iterator();
    while (rsite.hasNext())
    {
      Integer sid = rsite.next();
      mds += "\nfmRSDate(" + sid + ")=\t "+ "afmd=" + fmRSDate.get(sid);
    }

    // maxCNs
    Iterator<Integer> itc = maxCNs.keySet().iterator();
    while (itc.hasNext())
    {
      Integer sid = itc.next();
    for (Integer sid : maxCNs.keySet()) {
      ChangeNumber cn = maxCNs.get(sid);
      mds += "\nmaxCNs(" + sid + ")= " + cn.toStringUI();
    }

    // LDAP data
    Iterator<Integer> lsite = LDAPStates.keySet().iterator();
    while (lsite.hasNext())
    {
      Integer sid = lsite.next();
    for (Integer sid : LDAPStates.keySet()) {
      ServerState ss = LDAPStates.get(sid);
      mds += "\nLSData(" + sid + ")=\t" + "state=[" + ss.toString()
      + "] afmd=" + this.getApproxFirstMissingDate(sid);
          + "] afmd=" + this.getApproxFirstMissingDate(sid);

      mds += " missingDelay=" + this.getApproxDelay(sid);

      mds +=" missingCount=" + missingChanges.get(sid);
      mds += " missingCount=" + missingChanges.get(sid);
    }

    // RS data
    rsite = RSStates.keySet().iterator();
    while (rsite.hasNext())
    {
      Integer sid = rsite.next();
    for (Integer sid : RSStates.keySet()) {
      ServerState ss = RSStates.get(sid);
      mds += "\nRSData(" + sid + ")=\t" + "state=[" + ss.toString()
      + "] missingCount=" + missingChangesRS.get(sid);
@@ -321,10 +285,7 @@
   */
  public void setMaxCNs(ServerState state)
  {
    Iterator<Integer> it = state.iterator();
    while (it.hasNext())
    {
      int sid = it.next();
    for (Integer sid : state) {
      ChangeNumber newCN = state.getMaxChangeNumber(sid);
      setMaxCN(sid, newCN);
    }
@@ -352,17 +313,6 @@
  }

  /**
   * Get the highest know change number of the LDAP server with the provided
   * serverId.
   * @param serverId The server ID.
   * @return The highest change number.
   */
  public ChangeNumber getMaxCN(int serverId)
  {
    return maxCNs.get(serverId);
  }

  /**
   * Get the state of the LDAP server with the provided serverId.
   * @param serverId The server ID.
   * @return The server state.
@@ -454,9 +404,7 @@
   */
  public long getRSApproxFirstMissingDate(int serverId)
  {
    Long res;
    if ((res = fmRSDate.get(serverId)) != null)
      return res;
    // For now, we do store RS first missing change date
    return 0;
  }
}