Performance changes to import. Mostly for issue 3161.

dugan

30.31.2008 abd513b18b6271c54c01c42d44453f9c5f241599

Performance changes to import. Mostly for issue 3161.

 opendj-sdk/opends/src/server/org/opends/server/backends/jeb/Index.java

@@ -318,14 +318,14 @@
   * @param txn A transaction.
   * @param key The key to add the set to.
   * @param importIdSet The set of import IDs.
   * @param data Database entry to reuse for read
   * @throws DatabaseException If an database error occurs.
   */
  public void insert(Transaction txn, DatabaseEntry key,
                     ImportIDSet importIdSet)
                     ImportIDSet importIdSet, DatabaseEntry data)
  throws DatabaseException {

    OperationStatus status;
    DatabaseEntry data = new DatabaseEntry();
      status = read(txn, key, data, LockMode.RMW);
      if(status == OperationStatus.SUCCESS) {
        ImportIDSet newImportIDSet = new IntegerImportIDSet();
@@ -348,24 +348,26 @@


  /**
   * Add the specified entry ID to the provided keys in the keyset.
   * Add the specified import ID set to the provided keys in the keyset.
   *
   * @param txn  A transaction.
   * @param importIDSet A import ID set to use.
   * @param keySet  The set containing the keys.
   * @param entryID The entry ID.
   * @param keyData A key database entry to use.
   * @param data A database entry to use for data.
   * @return <CODE>True</CODE> if the insert was successful.
   * @throws DatabaseException If a database error occurs.
   */
  public synchronized
  boolean insert(Transaction txn, Set<byte[]> keySet, EntryID entryID)
  throws DatabaseException {
  boolean insert(Transaction txn, ImportIDSet importIDSet, Set<byte[]> keySet,
                 DatabaseEntry keyData, DatabaseEntry data)
          throws DatabaseException {
    for(byte[] key : keySet) {
      if(insertIDWithRMW(txn, new DatabaseEntry(key), new DatabaseEntry(),
              entryID.getDatabaseEntry(), entryID) !=
              OperationStatus.SUCCESS)  {
        return false;
      }
      keyData.setData(key);
      insert(txn, keyData, importIDSet, data);
    }
    keyData.setData(null);
    data.setData(null);
    return true;
  }


 opendj-sdk/opends/src/server/org/opends/server/backends/jeb/importLDIF/BufferManager.java

@@ -39,6 +39,7 @@
import org.opends.messages.Message;
import static org.opends.messages.JebMessages.*;
import java.util.*;
import java.util.concurrent.locks.ReentrantLock;


/**
@@ -70,11 +71,28 @@
  private TreeMap<KeyHashElement, KeyHashElement> elementMap =
                        new TreeMap<KeyHashElement, KeyHashElement>();

  //The current backup map being used.
  private int currentMap = 1;

  //Reference to use when the maps are switched.
  private TreeMap<KeyHashElement, KeyHashElement> backupMap;

  //The two backup maps to insert into if the main element map is being used.
  private TreeMap<KeyHashElement, KeyHashElement> backupMap2 =
                        new TreeMap<KeyHashElement, KeyHashElement>();
  private TreeMap<KeyHashElement, KeyHashElement> backupMap1 =
                        new TreeMap<KeyHashElement, KeyHashElement>();

  //Overhead values determined from using JHAT. They appear to be the same
  //for both 32 and 64 bit machines. Close enough.
  private final static int TREEMAP_ENTRY_OVERHEAD = 29;
  private final static int KEY_ELEMENT_OVERHEAD = 32;

  //Lock used to get main element map.
  private ReentrantLock lock = new ReentrantLock();

  //Object to synchronize on if backup maps are being written.
  private Object backupSynchObj = new Object();

  /**
   * Create buffer manager instance.
@@ -85,6 +103,7 @@
  public BufferManager(long memoryLimit, int importThreadCount) {
    this.memoryLimit = memoryLimit;
    this.nextElem = null;
    this.backupMap = backupMap1;
  }

  /**
@@ -95,45 +114,128 @@
   * @param entry The entry used to build the key set.
   * @param entryID The entry ID to insert into the key set.
   * @param txn A transaction.
   * @param keySet Keyset hash to store the keys in.
   * @throws DatabaseException If a problem happened during a flushAll cycle.
   */

  void insert(Index index, Entry entry,
              EntryID entryID, Transaction txn)
              EntryID entryID, Transaction txn, Set<byte[]> keySet)
          throws DatabaseException {

    Set<byte[]> keySet = new HashSet<byte[]>();
    keySet.clear();
    index.indexer.indexEntry(entry, keySet);
    synchronized(elementMap) {
      insertKeySet(keySet, index, entryID);
      //If over the memory limit and import hasn't completed
      //flush some keys from the cache to make room.
      if(memoryUsage > memoryLimit) {
        flushUntilUnderLimit();
      }
    if(!lock.tryLock()) {
      insertBackupMap(keySet, index, entryID);
      return;
    }
    insertKeySet(keySet, index, entryID, elementMap, true);
    if(!backupMap.isEmpty()) {
       mergeMap();
    }
    //If over the memory limit, flush some keys from the cache to make room.
    if(memoryUsage > memoryLimit) {
      flushUntilUnderLimit();
    }
    lock.unlock();
  }

  /**
   * Special case for id2children and id2subtree.
   * Insert an entry ID into the buffer using the both the specified index and
   * entry to build a key set.
   * @param id2children  The id2children index.
   * @param id2subtree The id2subtree index.
   * @param entry  The entry to used to build the keyset.
   * Insert an entry ID into buffer using specified id2children and id2subtree
   * indexes.
   *
   * @param id2children The id2children index to use.
   * @param id2subtree The id2subtree index to use.
   * @param entry The entry used to build the key set.
   * @param entryID The entry ID to insert into the key set.
   * @param txn A transaction.
   * @throws DatabaseException  If a problem happens formating the keyset.
   * @param txn  A transaction.
   * @param childKeySet id2children key set hash to use.
   * @param subKeySet subtree key set hash to use.
   * @throws DatabaseException If a problem occurs during processing.
   */
  void insert(Index id2children, Index id2subtree, Entry entry,
         EntryID entryID, Transaction txn) throws DatabaseException {
    Set<byte[]> childKeySet = new HashSet<byte[]>();
              EntryID entryID, Transaction txn, Set<byte[]> childKeySet,
              Set<byte[]> subKeySet) throws DatabaseException {
    childKeySet.clear();
    id2children.indexer.indexEntry(entry, childKeySet);
    Set<byte[]> subKeySet = new HashSet<byte[]>();
    subKeySet.clear();
    id2subtree.indexer.indexEntry(entry, subKeySet);
     synchronized(elementMap) {
      insertKeySet(childKeySet, id2children, entryID);
      insertKeySet(subKeySet, id2subtree, entryID);
    if(!lock.tryLock()) {
      insertBackupMap(childKeySet, id2children, subKeySet, id2subtree, entryID);
      return;
    }
    insertKeySet(childKeySet, id2children, entryID, elementMap, true);
    insertKeySet(subKeySet, id2subtree, entryID, elementMap, true);
    lock.unlock();
  }

  /**
   * Insert into a backup tree if can't get a lock on the main table.
   * @param childrenKeySet The id2children keyset to use.
   * @param id2children The id2children index to use.
   * @param subtreeKeySet The subtree keyset to use.
   * @param id2subtree The id2subtree index to use.
   * @param entryID The entry ID to insert into the key set.
   */
  void insertBackupMap(Set<byte[]> childrenKeySet, Index id2children,
                    Set<byte[]> subtreeKeySet,
                    Index id2subtree, EntryID entryID) {
    synchronized(backupSynchObj) {
      insertKeySet(childrenKeySet, id2children, entryID, backupMap,  false);
      insertKeySet(subtreeKeySet, id2subtree, entryID, backupMap,  false);
    }
  }


  /**
   * Insert specified keyset, index and entry ID into the backup map.
   *
   * @param keySet The keyset to use.
   * @param index The index to use.
   * @param entryID The entry ID to use.
   */
  void insertBackupMap(Set<byte[]> keySet, Index index, EntryID entryID) {
       synchronized(backupSynchObj) {
         insertKeySet(keySet, index, entryID, backupMap,  false);
    }
  }


  /**
   * Merge the backup map with the element map after switching the backup
   * map reference to an empty map.
   */
  void mergeMap() {
    TreeMap<KeyHashElement, KeyHashElement> tmpMap;
    synchronized(backupSynchObj) {
      tmpMap = backupMap;
      if(currentMap == 1) {
         backupMap = backupMap2;
         tmpMap = backupMap1;
         currentMap = 2;
      } else {
         backupMap = backupMap1;
         tmpMap = backupMap2;
         currentMap = 1;
      }
    }
    TreeSet<KeyHashElement>  tSet =
            new TreeSet<KeyHashElement>(tmpMap.keySet());
    for (KeyHashElement elem : tSet) {
      total++;
      if(!elementMap.containsKey(elem)) {
        elementMap.put(elem, elem);
        memoryUsage += TREEMAP_ENTRY_OVERHEAD + elem.getMemorySize();
      } else {
        KeyHashElement curElem = elementMap.get(elem);
        if(curElem.isDefined() || curElem.getIndex().getMaintainCount()) {
          int oldSize = curElem.getMemorySize();
          curElem.merge(elem);
          memoryUsage += (curElem.getMemorySize() - oldSize);
          hit++;
        }
      }
    }
    tmpMap.clear();
  }

  /**
@@ -141,27 +243,37 @@
   * @param keySet The key set to add to the map.
   * @param index  The index that eventually will contain the entry IDs.
   * @param entryID The entry ID to add to the entry ID set.
   * @param map The map to add the keys to
   * @param trackStats <CODE>True</CODE> if memory and usage should be tracked.
   */
  private void insertKeySet(Set<byte[]> keySet, Index index, EntryID entryID) {
      int entryLimit = index.getIndexEntryLimit();
      for(byte[] key : keySet) {
         KeyHashElement elem = new KeyHashElement(key, index, entryID);
         total++;
         if(!elementMap.containsKey(elem)) {
            elementMap.put(elem, elem);
            memoryUsage += TREEMAP_ENTRY_OVERHEAD + elem.getMemorySize();
         } else {
           KeyHashElement curElem = elementMap.get(elem);
           if(curElem.isDefined() || index.getMaintainCount()) {
            int oldSize = curElem.getMemorySize();
            curElem.addEntryID(entryID, entryLimit);
            int newSize = curElem.getMemorySize();
            //Might have moved from defined to undefined.
            memoryUsage += (newSize - oldSize);
            hit++;
           }
         }
  private void insertKeySet(Set<byte[]> keySet, Index index, EntryID entryID,
                            TreeMap<KeyHashElement, KeyHashElement> map,
                            boolean trackStats) {
    KeyHashElement elem = new KeyHashElement();
    int entryLimit = index.getIndexEntryLimit();
    for(byte[] key : keySet) {
      elem.reset(key, index);
      if(trackStats) {
        total++;
      }
      if(!map.containsKey(elem)) {
        KeyHashElement newElem = new KeyHashElement(key, index, entryID);
        map.put(newElem, newElem);
        if(trackStats) {
          memoryUsage += TREEMAP_ENTRY_OVERHEAD + newElem.getMemorySize();
        }
      } else {
        KeyHashElement curElem = map.get(elem);
        if(curElem.isDefined() || index.getMaintainCount()) {
          int oldSize = curElem.getMemorySize();
          curElem.addEntryID(entryID, entryLimit);
          if(trackStats) {
            memoryUsage += (curElem.getMemorySize() - oldSize);
            hit++;
          }
        }
      }
    }
  }

  /**
@@ -176,6 +288,8 @@
    } else {
      iter = elementMap.tailMap(nextElem).keySet().iterator();
    }
    DatabaseEntry dbEntry = new DatabaseEntry();
    DatabaseEntry entry = new DatabaseEntry();
    while((memoryUsage + extraBytes) > memoryLimit) {
      if(iter.hasNext()) {
        KeyHashElement curElem = iter.next();
@@ -183,8 +297,8 @@
        if(curElem.isDefined()) {
          int oldSize = curElem.getMemorySize();
          Index index = curElem.getIndex();
          index.insert(null, new DatabaseEntry(curElem.getKey()),
                  curElem.getIDSet());
          dbEntry.setData(curElem.getKey());
          index.insert(null, dbEntry, curElem.getIDSet(), entry);
          if(curElem.isDefined()) {
             memoryUsage -= TREEMAP_ENTRY_OVERHEAD + curElem.getMemorySize();
             iter.remove();
@@ -222,12 +336,15 @@
   * @throws DatabaseException If an error occurred during the insert.
   */
  void flushAll() throws DatabaseException {
    mergeMap();
    TreeSet<KeyHashElement>  tSet =
            new TreeSet<KeyHashElement>(elementMap.keySet());
    DatabaseEntry dbEntry = new DatabaseEntry();
    DatabaseEntry entry = new DatabaseEntry();
    for (KeyHashElement curElem : tSet) {
      Index index = curElem.getIndex();
      index.insert(null, new DatabaseEntry(curElem.getKey()),
              curElem.getIDSet());
      dbEntry.setData(curElem.getKey());
      index.insert(null, dbEntry, curElem.getIDSet(), entry);
    }
  }

@@ -253,6 +370,27 @@
    private int keyHashCode;

    /**
     * Empty constructor for use when the element is being reused.
     */
    public KeyHashElement() {}

    /**
     * Reset the element. Used when the element is being reused.
     *
     * @param key The new key to reset to.
     * @param index The new index to reset to.
     */
    public void reset(byte[] key, Index index) {
      this.key = key;
      this.index = index;
      this.indexHashCode = System.identityHashCode(index);
      this.keyHashCode = Arrays.hashCode(key);
      if(this.importIDSet != null) {
        this.importIDSet.reset();
      }
    }

    /**
     * Create instance of an element for the specified key and index, the add
     * the specified entry ID to the ID set.
     *
@@ -390,7 +528,6 @@
        throw new NullPointerException();
      }
      KeyHashElement inElem = (KeyHashElement) o;
    //  int keyCompare = compare(key, inElem.key);
      int keyCompare = compare(inElem);
      if(keyCompare == 0) {
        if(indexHashCode == inElem.indexHashCode) {
@@ -414,5 +551,14 @@
              MemoryBudget.byteArraySize(key.length) +
              importIDSet.getMemorySize();
    }

    /**
     * Merge the specified element with this element.
     * @param e The element to merge.
     */
    public void merge(KeyHashElement e) {
      importIDSet.merge(e.importIDSet, e.getIndex().getIndexEntryLimit(),
              e.getIndex().getMaintainCount());
    }
  }
}

 opendj-sdk/opends/src/server/org/opends/server/backends/jeb/importLDIF/ImportIDSet.java

@@ -85,15 +85,37 @@
                       int entryLimit, boolean maintainCount);

  /**
   * Merge the specified import ID set with the current import ID set using the
   * specified entry limit an maintain count values.
   *
   * @param bufImportIDSet The import ID set to merge.
   * @param entryLimit The entry limit to use.
   * @param maintainCount <CODE>True</CODE> if maintain count is being kept.
   */
  public void
  merge(ImportIDSet bufImportIDSet, int entryLimit, boolean maintainCount);

  /**
   * Set the import ID set to the undefined state.
   */
  public void setUndefined();


  /**
   * Return the undefined size.
   *
   * @return The undefined count.
   */
  public long getUndefinedSize();

  /**
   * Reset set.
   */
  public void reset();

  /**
   * Set the first entry ID to the specified entry ID.
   *
   * @param entryID The entry ID to use.
   */
  public void setEntryID(EntryID entryID);
}

 opendj-sdk/opends/src/server/org/opends/server/backends/jeb/importLDIF/IntegerImportIDSet.java

@@ -74,6 +74,26 @@
    count=1;
  }

  /**
   * {@inheritDoc}
   */
  public void setEntryID(EntryID id) {
    if(array == null) {
      this.array = new int[1];
    }
    reset();
    this.array[0] = (int) id.longValue();
    count=1;
  }

  /**
   * {@inheritDoc}
   */
  public void reset() {
    count = 0;
    isDefined = true;
    undefinedSize = 0;
  }

  /**
   * {@inheritDoc}
@@ -109,6 +129,35 @@
    }
  }

  /**
   * {@inheritDoc}
   */
  public void
  merge(ImportIDSet importIDSet, int limit, boolean maintainCount) {
    if(!isDefined()) {
      if(maintainCount)  {
        if(importIDSet.isDefined()) {
          undefinedSize += importIDSet.size();
        } else {
          undefinedSize += importIDSet.getUndefinedSize();
        }
      }
      return;
    }
    if(isDefined() && ((count + importIDSet.size()) > limit)) {
      isDefined = false;
      if(maintainCount)  {
        undefinedSize = size() + importIDSet.size();
      } else {
        undefinedSize = Long.MAX_VALUE;
      }
      array = null;
      count = 0;
    } else {
      addAll((IntegerImportIDSet) importIDSet);
    }
  }


  /**
   * {@inheritDoc}

 opendj-sdk/opends/src/server/org/opends/server/backends/jeb/importLDIF/LongImportIDSet.java

@@ -88,6 +88,27 @@
     count=1;
   }

  /**
   * {@inheritDoc}
   */
  public void setEntryID(EntryID id) {
    if(array == null) {
      this.array = new long[1];
    }
    reset();
    this.array[0] = id.longValue();
    count=1;
  }


    /**
   * {@inheritDoc}
   */
  public void reset() {
    count = 0;
    isDefined = true;
    undefinedSize = 0;
  }

  /**
   * {@inheritDoc}
@@ -122,6 +143,35 @@
    }
  }

  /**
   * {@inheritDoc}
   */
  public void
  merge(ImportIDSet importIDSet, int limit, boolean maintainCount) {
    if(!isDefined()) {
      if(maintainCount)  {
        if(importIDSet.isDefined()) {
          undefinedSize += importIDSet.size();
        } else {
          undefinedSize += importIDSet.getUndefinedSize();
        }
      }
      return;
    }
    if(isDefined() && ((count + importIDSet.size()) > limit)) {
      isDefined = false;
      if(maintainCount)  {
        undefinedSize = size() + importIDSet.size();
      } else {
        undefinedSize = Long.MAX_VALUE;
      }
      array = null;
      count = 0;
    } else {
      addAll((LongImportIDSet) importIDSet);
    }
  }


  /**
   * {@inheritDoc}

 opendj-sdk/opends/src/server/org/opends/server/backends/jeb/importLDIF/WorkThread.java

@@ -40,6 +40,7 @@
import com.sleepycat.je.DatabaseException;
import com.sleepycat.je.Transaction;
import com.sleepycat.je.LockMode;
import com.sleepycat.je.DatabaseEntry;

/**
 * A thread to process import entries from a queue.  Multiple instances of
@@ -77,6 +78,15 @@
  //The substring buffer manager to use.
  private BufferManager bufferMgr;

  //These are used to try and keep memory usage down.
  private Set<byte[]> insertKeySet = new HashSet<byte[]>();
  private Set<byte[]> childKeySet = new HashSet<byte[]>();
  private Set<byte[]> subtreeKeySet = new HashSet<byte[]>();
  private Set<byte[]> delKeySet = new HashSet<byte[]>();
  private DatabaseEntry keyData = new DatabaseEntry();
  private DatabaseEntry data = new DatabaseEntry();
  ImportIDSet importIDSet = new IntegerImportIDSet();

  /**
   * Create a work thread instance using the specified parameters.
   *
@@ -239,7 +249,7 @@
          insert(index, entry, entryID, txn);
        }
        if((index=attributeIndex.getSubstringIndex()) != null) {
          bufferMgr.insert(index,entry, entryID, txn);
          bufferMgr.insert(index,entry, entryID, txn, insertKeySet);
        }
        if((index=attributeIndex.getOrderingIndex()) != null) {
          insert(index, entry, entryID, txn);
@@ -271,7 +281,8 @@
    }
    Index id2children = context.getEntryContainer().getID2Children();
    Index id2subtree = context.getEntryContainer().getID2Subtree();
    bufferMgr.insert(id2children, id2subtree, entry, entryID, txn);
    bufferMgr.insert(id2children, id2subtree, entry, entryID, txn,
                    childKeySet, subtreeKeySet);
  }

  /**
@@ -288,9 +299,10 @@
  private boolean
  insert(Index index, Entry entry, EntryID entryID,
         Transaction txn) throws DatabaseException {
    Set<byte[]> keySet = new HashSet<byte[]>();
    index.indexer.indexEntry(entry, keySet);
    return index.insert(txn, keySet,  entryID);
    insertKeySet.clear();
    index.indexer.indexEntry(entry, insertKeySet);
    importIDSet.setEntryID(entryID);
    return index.insert(txn, importIDSet, insertKeySet, keyData, data);
  }

  /**
@@ -306,9 +318,9 @@
  private void
  delete(Index index, Entry entry, EntryID entryID,
         Transaction txn) throws DatabaseException {
    Set<byte[]> keySet = new HashSet<byte[]>();
    index.indexer.indexEntry(entry, keySet);
    index.delete(txn, keySet,  entryID);
    delKeySet.clear();
    index.indexer.indexEntry(entry, delKeySet);
    index.delete(txn, delKeySet,  entryID);
  }

  /**