From 0c04328d7c86a6b2d4badb96c79a22aba2fa7eca Mon Sep 17 00:00:00 2001
From: dugan <dugan@localhost>
Date: Fri, 16 May 2008 14:18:26 +0000
Subject: [PATCH] Fixes to  improve performance using default JVM ergonomics:

---
 opends/src/server/org/opends/server/backends/jeb/importLDIF/ImportIDSet.java        |   15 ++
 opends/src/server/org/opends/server/backends/jeb/EntryContainer.java                |   23 ++
 opends/src/server/org/opends/server/backends/jeb/importLDIF/WorkThread.java         |  122 ++++++++--------
 opends/src/server/org/opends/server/backends/jeb/importLDIF/IntegerImportIDSet.java |   21 +++
 opends/src/server/org/opends/server/backends/jeb/importLDIF/Importer.java           |   21 ++
 opends/src/server/org/opends/server/backends/jeb/importLDIF/LongImportIDSet.java    |   20 ++
 opends/src/server/org/opends/server/backends/jeb/importLDIF/DNContext.java          |    8 
 opends/src/server/org/opends/server/backends/jeb/AttributeIndex.java                |   32 ++++
 opends/src/server/org/opends/server/backends/jeb/importLDIF/BufferManager.java      |   33 ++-
 opends/src/server/org/opends/server/backends/jeb/Index.java                         |   97 +++++++++----
 10 files changed, 274 insertions(+), 118 deletions(-)

diff --git a/opends/src/server/org/opends/server/backends/jeb/AttributeIndex.java b/opends/src/server/org/opends/server/backends/jeb/AttributeIndex.java
index 5795edf..2af0d46 100644
--- a/opends/src/server/org/opends/server/backends/jeb/AttributeIndex.java
+++ b/opends/src/server/org/opends/server/backends/jeb/AttributeIndex.java
@@ -1224,6 +1224,38 @@
   }
 
   /**
+   * Close cursors related to the attribute indexes.
+   *
+   * @throws DatabaseException If a database error occurs.
+   */
+  public void closeCursors() throws DatabaseException {
+    if (equalityIndex != null)
+    {
+      equalityIndex.closeCursor();
+    }
+
+    if (presenceIndex != null)
+    {
+      presenceIndex.closeCursor();
+    }
+
+    if (substringIndex != null)
+    {
+      substringIndex.closeCursor();
+    }
+
+    if (orderingIndex != null)
+    {
+      orderingIndex.closeCursor();
+    }
+
+    if (approximateIndex != null)
+    {
+      approximateIndex.closeCursor();
+    }
+  }
+
+  /**
    * Return the number of values that have exceeded the entry limit since this
    * object was created.
    *
diff --git a/opends/src/server/org/opends/server/backends/jeb/EntryContainer.java b/opends/src/server/org/opends/server/backends/jeb/EntryContainer.java
index 903e8cd..ca6dc3e 100644
--- a/opends/src/server/org/opends/server/backends/jeb/EntryContainer.java
+++ b/opends/src/server/org/opends/server/backends/jeb/EntryContainer.java
@@ -1894,8 +1894,7 @@
      */
     public Transaction beginOperationTransaction() throws DatabaseException
     {
-      Transaction txn =  beginTransaction();
-      return txn;
+      return beginTransaction();
     }
 
     /**
@@ -2860,8 +2859,7 @@
      */
     public Transaction beginOperationTransaction() throws DatabaseException
     {
-      Transaction txn =  beginTransaction();
-      return txn;
+      return beginTransaction();
     }
 
     /**
@@ -3880,7 +3878,22 @@
   }
 
   /**
-   * Get a list of the databases opened by this entryContainer.
+   * Close cursors in the indexes of the context.
+   *
+   * @throws DatabaseException If a database error occurs.
+   */
+  public void closeIndexCursors() throws DatabaseException {
+    id2children.closeCursor();
+    id2subtree.closeCursor();
+    for (AttributeIndex index : attrIndexMap.values())
+    {
+      index.closeCursors();
+    }
+  }
+
+
+  /**
+   * Get a list of the databases opened by the entryContainer.
    * @param dbList A list of database containers.
    */
   public void listDatabases(List<DatabaseContainer> dbList)
diff --git a/opends/src/server/org/opends/server/backends/jeb/Index.java b/opends/src/server/org/opends/server/backends/jeb/Index.java
index b9874cb..b8e1c71 100644
--- a/opends/src/server/org/opends/server/backends/jeb/Index.java
+++ b/opends/src/server/org/opends/server/backends/jeb/Index.java
@@ -114,6 +114,9 @@
    */
   private boolean rebuildRunning = false;
 
+  //Thread local area to store per thread cursors.
+  private ThreadLocal<Cursor> curLocal = new ThreadLocal<Cursor>();
+
 
   /**
    * Create a new index object.
@@ -312,45 +315,66 @@
 
 
   /**
-   * Add the specified import ID set to the provided key. Used during
-   * substring buffer flushing.
+   * Insert the specified import ID set into this index a the provided key.
    *
-   * @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
+   * @param data Database entry to reuse for read.
+   * @param cursor A database cursor to use.
    * @throws DatabaseException If an database error occurs.
    */
-  public void insert(Transaction txn, DatabaseEntry key,
-                     ImportIDSet importIdSet, DatabaseEntry data)
-  throws DatabaseException {
 
-    OperationStatus status;
-      status = read(txn, key, data, LockMode.RMW);
-      if(status == OperationStatus.SUCCESS) {
-        ImportIDSet newImportIDSet = new IntegerImportIDSet();
-        if (newImportIDSet.merge(data.getData(), importIdSet,
-                                 indexEntryLimit, maintainCount)) {
-          entryLimitExceededCount++;
-        }
-        data.setData(newImportIDSet.toDatabase());
-      } else if(status == OperationStatus.NOTFOUND) {
-        if(!importIdSet.isDefined()) {
-          entryLimitExceededCount++;
-        }
-        data.setData(importIdSet.toDatabase());
-      } else {
-        //Should never happen during import.
-        throw new DatabaseException();
+  private void
+  insert(DatabaseEntry key, ImportIDSet importIdSet,
+         DatabaseEntry data, Cursor cursor) throws DatabaseException {
+    OperationStatus status =
+            cursor.getSearchKey(key, data, LockMode.DEFAULT);
+    if(status == OperationStatus.SUCCESS) {
+      ImportIDSet newImportIDSet = new IntegerImportIDSet();
+      if (newImportIDSet.merge(data.getData(), importIdSet,
+              indexEntryLimit, maintainCount) && importIdSet.isDirty()) {
+        entryLimitExceededCount++;
+        importIdSet.setDirty(false);
       }
-      put(txn,key, data);
+      data.setData(newImportIDSet.toDatabase());
+      cursor.putCurrent(data);
+    } else if(status == OperationStatus.NOTFOUND) {
+      if(!importIdSet.isDefined() && importIdSet.isDirty()) {
+        entryLimitExceededCount++;
+        importIdSet.setDirty(false);
+      }
+      data.setData(importIdSet.toDatabase());
+      cursor.put(key,data);
+    } else {
+      //Should never happen during import.
+      throw new DatabaseException();
+    }
+  }
+
+  /**
+   * Insert the specified import ID set into this index. Creates a DB
+   * cursor if needed.
+   *
+   * @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 a database error occurs.
+   */
+  public void
+  insert(DatabaseEntry key, ImportIDSet importIdSet,
+         DatabaseEntry data) throws DatabaseException {
+    Cursor cursor = curLocal.get();
+    if(cursor == null) {
+      cursor = openCursor(null, null);
+      curLocal.set(cursor);
+    }
+    insert(key, importIdSet, data, cursor);
   }
 
 
   /**
    * 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 keyData A key database entry to use.
@@ -358,13 +382,19 @@
    * @return <CODE>True</CODE> if the insert was successful.
    * @throws DatabaseException If a database error occurs.
    */
+
   public synchronized
-  boolean insert(Transaction txn, ImportIDSet importIDSet, Set<byte[]> keySet,
+  boolean insert(ImportIDSet importIDSet, Set<byte[]> keySet,
                  DatabaseEntry keyData, DatabaseEntry data)
           throws DatabaseException {
+    Cursor cursor = curLocal.get();
+    if(cursor == null) {
+      cursor = openCursor(null, null);
+      curLocal.set(cursor);
+    }
     for(byte[] key : keySet) {
       keyData.setData(key);
-      insert(txn, keyData, importIDSet, data);
+      insert(keyData, importIDSet, data, cursor);
     }
     keyData.setData(null);
     data.setData(null);
@@ -1131,6 +1161,17 @@
   }
 
   /**
+   * Close any cursors open against this index.
+   *
+   * @throws DatabaseException  If a database error occurs.
+   */
+  public void closeCursor() throws DatabaseException {
+    Cursor cursor = curLocal.get();
+    if(cursor != null) {
+      cursor.close();
+    }
+  }
+  /**
    * Increment the count of the number of keys that have exceeded the entry
    * limit since this object was created.
    */
diff --git a/opends/src/server/org/opends/server/backends/jeb/importLDIF/BufferManager.java b/opends/src/server/org/opends/server/backends/jeb/importLDIF/BufferManager.java
index a1a8f74..b314d93 100644
--- a/opends/src/server/org/opends/server/backends/jeb/importLDIF/BufferManager.java
+++ b/opends/src/server/org/opends/server/backends/jeb/importLDIF/BufferManager.java
@@ -31,7 +31,6 @@
 import org.opends.server.types.Entry;
 import org.opends.server.backends.jeb.Index;
 import org.opends.server.backends.jeb.EntryID;
-import com.sleepycat.je.Transaction;
 import com.sleepycat.je.DatabaseException;
 import com.sleepycat.je.DatabaseEntry;
 import com.sleepycat.je.dbi.MemoryBudget;
@@ -92,15 +91,14 @@
   private ReentrantLock lock = new ReentrantLock();
 
   //Object to synchronize on if backup maps are being written.
-  private Object backupSynchObj = new Object();
+  private final Object backupSynchObj = new Object();
 
   /**
    * Create buffer manager instance.
    *
    * @param memoryLimit The memory limit.
-   * @param importThreadCount  The count of import worker threads.
    */
-  public BufferManager(long memoryLimit, int importThreadCount) {
+  public BufferManager(long memoryLimit) {
     this.memoryLimit = memoryLimit;
     this.nextElem = null;
     this.backupMap = backupMap1;
@@ -113,13 +111,12 @@
    * @param index  The 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.
    * @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, Set<byte[]> keySet)
+              EntryID entryID, Set<byte[]> keySet)
           throws DatabaseException {
 
     keySet.clear();
@@ -147,13 +144,12 @@
    * @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.
    * @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, Set<byte[]> childKeySet,
+              EntryID entryID, Set<byte[]> childKeySet,
               Set<byte[]> subKeySet) throws DatabaseException {
     childKeySet.clear();
     id2children.indexer.indexEntry(entry, childKeySet);
@@ -207,7 +203,6 @@
   void mergeMap() {
     TreeMap<KeyHashElement, KeyHashElement> tmpMap;
     synchronized(backupSynchObj) {
-      tmpMap = backupMap;
       if(currentMap == 1) {
          backupMap = backupMap2;
          tmpMap = backupMap1;
@@ -298,7 +293,7 @@
           int oldSize = curElem.getMemorySize();
           Index index = curElem.getIndex();
           dbEntry.setData(curElem.getKey());
-          index.insert(null, dbEntry, curElem.getIDSet(), entry);
+          index.insert(dbEntry, curElem.getIDSet(), entry);
           if(curElem.isDefined()) {
              memoryUsage -= TREEMAP_ENTRY_OVERHEAD + curElem.getMemorySize();
              iter.remove();
@@ -342,9 +337,11 @@
     DatabaseEntry dbEntry = new DatabaseEntry();
     DatabaseEntry entry = new DatabaseEntry();
     for (KeyHashElement curElem : tSet) {
-      Index index = curElem.getIndex();
-      dbEntry.setData(curElem.getKey());
-      index.insert(null, dbEntry, curElem.getIDSet(), entry);
+      if(curElem.isDirty()) {
+        Index index = curElem.getIndex();
+        dbEntry.setData(curElem.getKey());
+        index.insert(dbEntry, curElem.getIDSet(), entry);
+      }
     }
   }
 
@@ -560,5 +557,15 @@
       importIDSet.merge(e.importIDSet, e.getIndex().getIndexEntryLimit(),
               e.getIndex().getMaintainCount());
     }
+
+    /**
+     * Return if an undefined import ID set has been written to the index DB.
+     *
+     * @return <CODE>True</CODE> if an undefined importID set has been written
+     * to the index DB.
+     */
+    public boolean isDirty() {
+      return importIDSet.isDirty();
+    }
   }
 }
diff --git a/opends/src/server/org/opends/server/backends/jeb/importLDIF/DNContext.java b/opends/src/server/org/opends/server/backends/jeb/importLDIF/DNContext.java
index 8f95554..2d27644 100644
--- a/opends/src/server/org/opends/server/backends/jeb/importLDIF/DNContext.java
+++ b/opends/src/server/org/opends/server/backends/jeb/importLDIF/DNContext.java
@@ -34,7 +34,6 @@
 import org.opends.server.backends.jeb.*;
 
 import com.sleepycat.je.DatabaseException;
-import com.sleepycat.je.Transaction;
 import com.sleepycat.je.LockMode;
 
 import java.util.concurrent.BlockingQueue;
@@ -116,7 +115,7 @@
 
   //Used to synchronize the parent ID map, since multiple worker threads
   //can be accessing it.
-  private Object synchObject = new Object();
+  private final Object synchObject = new Object();
 
   /**
    * The number of LDAP entries added to the database, used to update the
@@ -441,12 +440,11 @@
      * Get the Entry ID of the parent entry.
      * @param parentDN  The parent DN.
      * @param dn2id The DN2ID DB.
-     * @param txn A database transaction,
      * @return The entry ID of the parent entry.
      * @throws DatabaseException If a DB error occurs.
      */
     public
-    EntryID getParentID(DN parentDN, DN2ID dn2id, Transaction txn)
+    EntryID getParentID(DN parentDN, DN2ID dn2id)
             throws DatabaseException {
       EntryID parentID;
       synchronized(synchObject) {
@@ -469,7 +467,7 @@
           return null;
         }
       }
-      parentID = dn2id.get(txn, parentDN, LockMode.DEFAULT);
+      parentID = dn2id.get(null, parentDN, LockMode.DEFAULT);
       //If the parent is in dn2id, add it to the cache.
       if (parentID != null) {
         synchronized(synchObject) {
diff --git a/opends/src/server/org/opends/server/backends/jeb/importLDIF/ImportIDSet.java b/opends/src/server/org/opends/server/backends/jeb/importLDIF/ImportIDSet.java
index fe8c7df..186fca0 100644
--- a/opends/src/server/org/opends/server/backends/jeb/importLDIF/ImportIDSet.java
+++ b/opends/src/server/org/opends/server/backends/jeb/importLDIF/ImportIDSet.java
@@ -118,4 +118,19 @@
    * @param entryID The entry ID to use.
    */
   public void setEntryID(EntryID entryID);
+
+  /**
+   * Return if a undefined entry ID set has been written to the index DB.
+   *
+   * @return Return <CODE>True</CODE>if the undefined entry ID set has been
+   * written to the index DB.
+   */
+  public boolean isDirty();
+
+  /**
+   * Set the dirty flag to the specifed value.
+   *
+   * @param dirty The value to set the flag to.
+   */
+  public void setDirty(boolean dirty);
 }
diff --git a/opends/src/server/org/opends/server/backends/jeb/importLDIF/Importer.java b/opends/src/server/org/opends/server/backends/jeb/importLDIF/Importer.java
index 8c9b3e2..9464a18 100644
--- a/opends/src/server/org/opends/server/backends/jeb/importLDIF/Importer.java
+++ b/opends/src/server/org/opends/server/backends/jeb/importLDIF/Importer.java
@@ -178,11 +178,11 @@
     // Create one set of worker threads/buffer managers for each base DN.
     for (DNContext context : importMap.values()) {
       BufferManager bufferManager =
-                        new BufferManager(memoryPerContext, importThreadCount);
+                        new BufferManager(memoryPerContext);
       context.setBufferManager(bufferManager);
       for (int i = 0; i < importThreadCount; i++) {
         WorkThread t = new WorkThread(context.getWorkQueue(), i,
-                bufferManager, rootContainer);
+                bufferManager, rootContainer, importMap);
         t.setUncaughtExceptionHandler(this);
         threads.add(t);
         t.start();
@@ -557,6 +557,15 @@
     msg = NOTE_JEB_IMPORT_LDIF_FINAL_CLEAN.get();
     //Run the cleaner.
     runCleaner(msg);
+    closeIndexCursors();
+  }
+
+
+   private void closeIndexCursors() throws DatabaseException {
+    for (DNContext ic : importMap.values())
+    {
+      ic.getEntryContainer().closeIndexCursors();
+    }
   }
 
   /**
@@ -776,11 +785,11 @@
     long maxMemory = runtime.maxMemory();
     long totMemory = runtime.totalMemory();
     long totFreeMemory = (freeMemory + (maxMemory - totMemory));
-    long dbCacheLimit = (totFreeMemory * 45) / 100;
-    //If there are now substring indexes defined, set the DB cache
-    //size to 60% and take a minimal substring buffer.
+    long dbCacheLimit = (totFreeMemory * 60) / 100;
+    //If there are no substring indexes defined, set the DB cache
+    //size to 75% and take a minimal substring buffer.
     if(!hasSubIndexes) {
-      dbCacheLimit = (totFreeMemory * 60) / 100;
+      dbCacheLimit = (totFreeMemory * 75) / 100;
     }
     dbCacheSizeStr = Long.toString(dbCacheLimit);
     totalAvailBufferMemory = (totFreeMemory * 10) / 100;
diff --git a/opends/src/server/org/opends/server/backends/jeb/importLDIF/IntegerImportIDSet.java b/opends/src/server/org/opends/server/backends/jeb/importLDIF/IntegerImportIDSet.java
index f553d07..8b5eb0a 100644
--- a/opends/src/server/org/opends/server/backends/jeb/importLDIF/IntegerImportIDSet.java
+++ b/opends/src/server/org/opends/server/backends/jeb/importLDIF/IntegerImportIDSet.java
@@ -35,6 +35,10 @@
  */
 public class IntegerImportIDSet implements ImportIDSet {
 
+
+  //Indicate if a undefined import set has been written to the index DB.
+  private boolean dirty = true;
+
   //Gleamed from JHAT. The same for 32/64 bit.
   private final static int THIS_OVERHEAD = 25;
 
@@ -93,11 +97,25 @@
     count = 0;
     isDefined = true;
     undefinedSize = 0;
+    dirty = true;
   }
 
   /**
    * {@inheritDoc}
    */
+  public void setDirty(boolean dirty) {
+    this.dirty = dirty;
+  }
+
+  /**
+   * {@inheritDoc}
+   */
+  public boolean isDirty() {
+    return dirty;
+  }
+  /**
+   * {@inheritDoc}
+   */
   public boolean isDefined() {
     return isDefined;
   }
@@ -239,9 +257,11 @@
       if(dbUndefined) {
         isDefined=false;
         importIdSet.setUndefined();
+        undefinedSize = Long.MAX_VALUE;
       } else if(!importIdSet.isDefined()) {
         isDefined=false;
         incrLimitCount=true;
+        undefinedSize = Long.MAX_VALUE;
       } else {
         array = JebFormat.intArrayFromDatabaseBytes(dBbytes);
         if(array.length + importIdSet.size() > limit) {
@@ -249,6 +269,7 @@
           incrLimitCount=true;
           count = 0;
           importIdSet.setUndefined();
+          undefinedSize = Long.MAX_VALUE;
         } else {
           count = array.length;
           addAll((IntegerImportIDSet) importIdSet);
diff --git a/opends/src/server/org/opends/server/backends/jeb/importLDIF/LongImportIDSet.java b/opends/src/server/org/opends/server/backends/jeb/importLDIF/LongImportIDSet.java
index ac7affa..863f875 100644
--- a/opends/src/server/org/opends/server/backends/jeb/importLDIF/LongImportIDSet.java
+++ b/opends/src/server/org/opends/server/backends/jeb/importLDIF/LongImportIDSet.java
@@ -37,6 +37,8 @@
  */
 public class LongImportIDSet implements ImportIDSet {
 
+  //Indicate if a undefined import set has been written to the index DB.
+  private boolean dirty = true;
 
   //Overhead values gleamed from JHAT.
   private final static int LONGS_OVERHEAD;
@@ -108,6 +110,21 @@
     count = 0;
     isDefined = true;
     undefinedSize = 0;
+    dirty = true;
+  }
+
+  /**
+   * {@inheritDoc}
+   */
+  public void setDirty(boolean dirty) {
+    this.dirty = dirty;
+  }
+
+  /**
+   * {@inheritDoc}
+   */
+  public boolean isDirty() {
+    return dirty;
   }
 
   /**
@@ -183,15 +200,18 @@
 
     if(dbUndefined) {
       isDefined=false;
+      undefinedSize = Long.MAX_VALUE;
     } else if(!importIdSet.isDefined()) {
       isDefined=false;
       incrLimitCount=true;
+      undefinedSize = Long.MAX_VALUE;
     } else {
       array = JebFormat.entryIDListFromDatabase(DBbytes);
       if(array.length + importIdSet.size() > limit) {
           isDefined=false;
           incrLimitCount=true;
           importIdSet.setUndefined();
+          undefinedSize = Long.MAX_VALUE;
       } else {
         count = array.length;
         addAll((LongImportIDSet) importIdSet);
diff --git a/opends/src/server/org/opends/server/backends/jeb/importLDIF/WorkThread.java b/opends/src/server/org/opends/server/backends/jeb/importLDIF/WorkThread.java
index 76b3af5..b690c8d 100644
--- a/opends/src/server/org/opends/server/backends/jeb/importLDIF/WorkThread.java
+++ b/opends/src/server/org/opends/server/backends/jeb/importLDIF/WorkThread.java
@@ -37,10 +37,8 @@
 import java.util.concurrent.BlockingQueue;
 import java.util.concurrent.TimeUnit;
 import java.util.*;
-import com.sleepycat.je.DatabaseException;
-import com.sleepycat.je.Transaction;
-import com.sleepycat.je.LockMode;
-import com.sleepycat.je.DatabaseEntry;
+
+import com.sleepycat.je.*;
 
 /**
  * A thread to process import entries from a queue.  Multiple instances of
@@ -72,8 +70,6 @@
    */
   private boolean stopRequested = false;
 
-  //The thread number related to a thread.
-  private int threadNumber;
 
   //The substring buffer manager to use.
   private BufferManager bufferMgr;
@@ -86,6 +82,7 @@
   private DatabaseEntry keyData = new DatabaseEntry();
   private DatabaseEntry data = new DatabaseEntry();
   ImportIDSet importIDSet = new IntegerImportIDSet();
+  private LinkedHashMap<DN, DNContext> importMap;
 
   /**
    * Create a work thread instance using the specified parameters.
@@ -94,15 +91,17 @@
    * @param threadNumber The thread number.
    * @param bufferMgr  The buffer manager to use.
    * @param rootContainer The root container.
+   * @param importMap The import map.
    */
   public WorkThread(BlockingQueue<WorkElement> workQueue, int threadNumber,
                                 BufferManager bufferMgr,
-                                RootContainer rootContainer) {
+                                RootContainer rootContainer,
+                                LinkedHashMap<DN, DNContext> importMap) {
     super("Import Worker Thread " + threadNumber);
-    this.threadNumber = threadNumber;
     this.workQueue = workQueue;
     this.bufferMgr = bufferMgr;
     this.rootContainer = rootContainer;
+    this.importMap = importMap;
   }
 
   /**
@@ -142,6 +141,7 @@
           }
         }
       } while (!stopRequested);
+      closeIndexCursors();
     } catch (Exception e) {
       if (debugEnabled()) {
         TRACER.debugCaught(DebugLogLevel.ERROR, e);
@@ -150,6 +150,19 @@
     }
   }
 
+
+  /**
+   * Close all database cursors opened by this thread.
+   *
+   * @throws DatabaseException If a database error occurs.
+   */
+  private void closeIndexCursors() throws DatabaseException {
+    for (DNContext ic : importMap.values())
+    {
+      ic.getEntryContainer().closeIndexCursors();
+    }
+  }
+
   /**
    * Process a work element.
    *
@@ -161,14 +174,13 @@
    */
   private void process(WorkElement element)
   throws DatabaseException, DirectoryException, JebException {
-    Transaction txn = null;
     EntryID entryID;
-    if((entryID = processDN2ID(element, txn)) == null)
+    if((entryID = processDN2ID(element)) == null)
       return;
-    if(!processID2Entry(element, entryID, txn))
+    if(!processID2Entry(element, entryID))
       return;
-    procesID2SCEntry(element, entryID, txn);
-    processIndexesEntry(element, entryID, txn);
+    procesID2SCEntry(element, entryID);
+    processIndexesEntry(element, entryID);
   }
 
   /**
@@ -178,12 +190,11 @@
    * @param element The work element.
    * @param existingEntry The existing entry to replace.
    * @param entryID The entry ID to remove from the keys.
-   * @param txn A transaction.
    * @throws DatabaseException If a database error occurs.
    */
   private void
   processIndexesEntryDelete(WorkElement element, Entry existingEntry,
-                            EntryID entryID, Transaction txn)
+                            EntryID entryID)
           throws DatabaseException {
     DNContext context = element.getContext();
     Map<AttributeType, AttributeIndex> attrIndexMap =
@@ -195,19 +206,19 @@
         AttributeIndex attributeIndex = mapEntry.getValue();
         Index index;
         if((index=attributeIndex.getEqualityIndex()) != null) {
-          delete(index, existingEntry, entryID, txn);
+          delete(index, existingEntry, entryID);
         }
         if((index=attributeIndex.getPresenceIndex()) != null) {
-          delete(index, existingEntry, entryID, txn);
+          delete(index, existingEntry, entryID);
         }
         if((index=attributeIndex.getSubstringIndex()) != null) {
-          delete(index, existingEntry, entryID, txn);
+          delete(index, existingEntry, entryID);
         }
         if((index=attributeIndex.getOrderingIndex()) != null) {
-          delete(index, existingEntry, entryID, txn);
+          delete(index, existingEntry, entryID);
         }
         if((index=attributeIndex.getApproximateIndex()) != null) {
-          delete(index, existingEntry, entryID, txn);
+          delete(index, existingEntry, entryID);
         }
       }
     }
@@ -218,11 +229,10 @@
    *
    * @param element The work element.
    * @param entryID The entry ID to process.
-   * @param txn A transaction.
    * @throws DatabaseException If an database error occurs.
    */
   private void
-  processIndexesEntry(WorkElement element, EntryID entryID, Transaction txn)
+  processIndexesEntry(WorkElement element, EntryID entryID)
           throws DatabaseException {
     Entry entry = element.getEntry();
     DNContext context = element.getContext();
@@ -231,7 +241,7 @@
             ldifImportConfig.replaceExistingEntries()) {
       Entry existingEntry = element.getExistingEntry();
       if(existingEntry != null) {
-          processIndexesEntryDelete(element, existingEntry, entryID, txn);
+          processIndexesEntryDelete(element, existingEntry, entryID);
       }
     }
     Map<AttributeType, AttributeIndex> attrIndexMap =
@@ -243,19 +253,19 @@
         AttributeIndex attributeIndex = mapEntry.getValue();
         Index index;
         if((index=attributeIndex.getEqualityIndex()) != null) {
-          insert(index, entry, entryID, txn);
+          insert(index, entry, entryID);
         }
         if((index=attributeIndex.getPresenceIndex()) != null) {
-          insert(index, entry, entryID, txn);
+          insert(index, entry, entryID);
         }
         if((index=attributeIndex.getSubstringIndex()) != null) {
-          bufferMgr.insert(index,entry, entryID, txn, insertKeySet);
+          bufferMgr.insert(index,entry, entryID, insertKeySet);
         }
         if((index=attributeIndex.getOrderingIndex()) != null) {
-          insert(index, entry, entryID, txn);
+          insert(index, entry, entryID);
         }
         if((index=attributeIndex.getApproximateIndex()) != null) {
-          insert(index, entry, entryID, txn);
+          insert(index, entry, entryID);
         }
       }
     }
@@ -266,12 +276,11 @@
    *
    * @param element The work element.
    * @param entryID The entry ID to process.
-   * @param txn A transaction.
    * @throws DatabaseException If an database error occurs.
    */
   private  void
-  procesID2SCEntry(WorkElement element, EntryID entryID,
-                   Transaction txn) throws DatabaseException {
+  procesID2SCEntry(WorkElement element, EntryID entryID)
+          throws DatabaseException {
     Entry entry = element.getEntry();
     DNContext context = element.getContext();
     LDIFImportConfig ldifImportConfig = context.getLDIFImportConfig();
@@ -281,7 +290,7 @@
     }
     Index id2children = context.getEntryContainer().getID2Children();
     Index id2subtree = context.getEntryContainer().getID2Subtree();
-    bufferMgr.insert(id2children, id2subtree, entry, entryID, txn,
+    bufferMgr.insert(id2children, id2subtree, entry, entryID,
                     childKeySet, subtreeKeySet);
   }
 
@@ -292,17 +301,15 @@
    * @param index  The index to insert into.
    * @param entry The entry to generate the keys from.
    * @param entryID The entry ID to insert.
-   * @param txn A transaction.
    * @return <CODE>True</CODE> if insert succeeded.
    * @throws DatabaseException If a database error occurs.
    */
   private boolean
-  insert(Index index, Entry entry, EntryID entryID,
-         Transaction txn) throws DatabaseException {
+  insert(Index index, Entry entry, EntryID entryID) throws DatabaseException {
     insertKeySet.clear();
     index.indexer.indexEntry(entry, insertKeySet);
     importIDSet.setEntryID(entryID);
-    return index.insert(txn, importIDSet, insertKeySet, keyData, data);
+    return index.insert(importIDSet, insertKeySet, keyData, data);
   }
 
   /**
@@ -312,15 +319,13 @@
    * @param index  The index to insert into.
    * @param entry The entry to generate the keys from.
    * @param entryID The entry ID to insert.
-   * @param txn A transaction.
    * @throws DatabaseException If a database error occurs.
    */
   private void
-  delete(Index index, Entry entry, EntryID entryID,
-         Transaction txn) throws DatabaseException {
+  delete(Index index, Entry entry, EntryID entryID) throws DatabaseException {
     delKeySet.clear();
     index.indexer.indexEntry(entry, delKeySet);
-    index.delete(txn, delKeySet,  entryID);
+    index.delete(null, delKeySet,  entryID);
   }
 
   /**
@@ -328,20 +333,19 @@
    *
    * @param element The work element containing the entry.
    * @param entryID The entry ID to use as the key.
-   * @param txn A transaction.
    * @return <CODE>True</CODE> If the insert succeeded.
    * @throws DatabaseException If a database error occurs.
    * @throws DirectoryException  If a directory error occurs.
    */
   private boolean
-  processID2Entry(WorkElement element, EntryID entryID, Transaction txn)
+  processID2Entry(WorkElement element, EntryID entryID)
           throws DatabaseException, DirectoryException {
     boolean ret;
     Entry entry = element.getEntry();
     DNContext context = element.getContext();
     ID2Entry id2entry = context.getEntryContainer().getID2Entry();
     DN2URI dn2uri = context.getEntryContainer().getDN2URI();
-    ret=id2entry.put(txn, entryID, entry);
+    ret=id2entry.put(null, entryID, entry);
     if(ret) {
       importedCount++;
       LDIFImportConfig ldifImportConfig = context.getLDIFImportConfig();
@@ -349,10 +353,10 @@
               ldifImportConfig.replaceExistingEntries()) {
         Entry existingEntry = element.getExistingEntry();
         if(existingEntry != null) {
-          dn2uri.replaceEntry(txn, existingEntry, entry);
+          dn2uri.replaceEntry(null, existingEntry, entry);
         }
       } else {
-        ret= dn2uri.addEntry(txn, entry);
+        ret= dn2uri.addEntry(null, entry);
       }
     }
     return ret;
@@ -362,13 +366,11 @@
    * Process entry from work element checking if it's parent exists.
    *
    * @param element The work element containing the entry.
-   * @param txn A transaction.
    * @return <CODE>True</CODE> If the insert succeeded.
    * @throws DatabaseException If a database error occurs.
    */
   private boolean
-  processParent(WorkElement element, Transaction txn)
-          throws DatabaseException {
+  processParent(WorkElement element) throws DatabaseException {
     Entry entry = element.getEntry();
     DNContext context = element.getContext();
     LDIFImportConfig ldifImportConfig = context.getLDIFImportConfig();
@@ -381,9 +383,9 @@
     DN parentDN = context.getEntryContainer().getParentWithinBase(entryDN);
     DN2ID dn2id = context.getEntryContainer().getDN2ID();
     if (parentDN != null) {
-      parentID = context.getParentID(parentDN, dn2id, txn);
+      parentID = context.getParentID(parentDN, dn2id);
       if (parentID == null) {
-        dn2id.remove(txn, entryDN);
+        dn2id.remove(null, entryDN);
         Message msg =
                 ERR_JEB_IMPORT_PARENT_NOT_FOUND.get(parentDN.toString());
         context.getLDIFReader().rejectLastEntry(msg);
@@ -406,7 +408,7 @@
         EntryContainer ec = context.getEntryContainer();
         for (DN dn = ec.getParentWithinBase(parentDN); dn != null;
              dn = ec.getParentWithinBase(dn)) {
-          if((nodeID =  getAncestorID(dn2id, dn, txn)) == null) {
+          if((nodeID =  getAncestorID(dn2id, dn)) == null) {
             return false;
           } else {
             IDs.add(nodeID);
@@ -420,12 +422,12 @@
     return true;
   }
 
-  private EntryID getAncestorID(DN2ID dn2id, DN dn, Transaction txn)
+  private EntryID getAncestorID(DN2ID dn2id, DN dn)
           throws DatabaseException {
     int i=0;
-    EntryID nodeID = dn2id.get(txn, dn, LockMode.DEFAULT);
+    EntryID nodeID = dn2id.get(null, dn, LockMode.DEFAULT);
     if(nodeID == null) {
-      while((nodeID = dn2id.get(txn, dn, LockMode.DEFAULT)) == null) {
+      while((nodeID = dn2id.get(null, dn, LockMode.DEFAULT)) == null) {
         try {
           Thread.sleep(50);
           if(i == 3) {
@@ -444,25 +446,23 @@
    * Process the a entry from the work element into the dn2id DB.
    *
    * @param element The work element containing the entry.
-   * @param txn A transaction.
    * @return An entry ID.
    * @throws DatabaseException If a database error occurs.
    * @throws JebException If a JEB error occurs.
    */
   private EntryID
-  processDN2ID(WorkElement element, Transaction txn)
-          throws DatabaseException, JebException {
+  processDN2ID(WorkElement element) throws DatabaseException, JebException {
     Entry entry = element.getEntry();
     DNContext context = element.getContext();
     DN2ID dn2id = context.getEntryContainer().getDN2ID();
     LDIFImportConfig ldifImportConfig = context.getLDIFImportConfig();
     DN entryDN = entry.getDN();
-    EntryID entryID = dn2id.get(txn, entryDN, LockMode.DEFAULT);
+    EntryID entryID = dn2id.get(null, entryDN, LockMode.DEFAULT);
     if (entryID != null) {
       if (ldifImportConfig.appendToExistingData() &&
               ldifImportConfig.replaceExistingEntries()) {
         ID2Entry id2entry = context.getEntryContainer().getID2Entry();
-        Entry existingEntry = id2entry.get(txn, entryID, LockMode.DEFAULT);
+        Entry existingEntry = id2entry.get(null, entryID, LockMode.DEFAULT);
         element.setExistingEntry(existingEntry);
       } else {
         Message msg = WARN_JEB_IMPORT_ENTRY_EXISTS.get();
@@ -470,7 +470,7 @@
         entryID = null;
       }
     } else {
-      if(!processParent(element, txn))
+      if(!processParent(element))
         return null;
       if (ldifImportConfig.appendToExistingData() &&
               ldifImportConfig.replaceExistingEntries()) {
@@ -479,7 +479,7 @@
         ArrayList IDs = (ArrayList)entry.getAttachment();
         entryID = (EntryID)IDs.get(0);
       }
-      dn2id.insert(txn, entryDN, entryID);
+      dn2id.insert(null, entryDN, entryID);
     }
     context.removePending(entryDN);
     return entryID;

--
Gitblit v1.10.0