Fix performance regression in Import, an issue with --append more and an is...

ludovicp

05.08.2010 06a2720e00f89a73a5617aabc4ee91cbac62fbee

Fix performance regression in Import, an issue with --append more and an issue with the way the DB config was used during import (Issue #4473).
Memory allocation for import is now based on the usable memory for offline mode and configure DB cache size for online.
Computation of usable JVM currently only works properly with Sun JVM for ParallelGC and CMSGC.

 opends/src/server/org/opends/server/backends/jeb/BackendImpl.java

@@ -31,6 +31,7 @@
import java.io.File;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.TimeUnit;

import java.io.FileInputStream;
import java.io.FilenameFilter;
@@ -39,6 +40,7 @@
import java.util.zip.CheckedInputStream;

import com.sleepycat.je.DatabaseException;
import com.sleepycat.je.Durability;
import com.sleepycat.je.EnvironmentConfig;
import com.sleepycat.je.EnvironmentFailureException;

@@ -1110,8 +1112,18 @@

    try
    {
      EnvironmentConfig envConfig =
              ConfigurableEnvironment.parseConfigEntry(cfg);
      EnvironmentConfig envConfig = new EnvironmentConfig();

      envConfig.setAllowCreate(true);
      envConfig.setTransactional(false);
      envConfig.setDurability(Durability.COMMIT_NO_SYNC);
      envConfig.setLockTimeout(0, TimeUnit.SECONDS);
      envConfig.setTxnTimeout(0, TimeUnit.SECONDS);
      envConfig.setConfigParam(EnvironmentConfig.CLEANER_MIN_FILE_UTILIZATION,
          String.valueOf(cfg.getDBCleanerMinUtilization()));
      envConfig.setConfigParam(EnvironmentConfig.LOG_FILE_MAX, String
          .valueOf(cfg.getDBLogFileMax()));

      if(!importConfig.appendToExistingData()) {
        if(importConfig.clearBackend() || cfg.getBaseDN().size() <= 1) {
          // We have the writer lock on the environment, now delete the
@@ -1126,15 +1138,8 @@
          }
        }
      }
      envConfig.setReadOnly(false);
      envConfig.setAllowCreate(true);
      envConfig.setTransactional(false);
      envConfig.setConfigParam(EnvironmentConfig.ENV_IS_LOCKING, "true");
      envConfig.setConfigParam(EnvironmentConfig.ENV_RUN_CHECKPOINTER, "false");
      envConfig.setConfigParam(EnvironmentConfig.ENV_RUN_CLEANER, "false");
      envConfig.setConfigParam(EnvironmentConfig.EVICTOR_LRU_ONLY, "false");
      envConfig.setConfigParam(EnvironmentConfig.EVICTOR_NODES_PER_SCAN, "128");
      Importer importer = Importer.getInstance(importConfig, cfg, envConfig);

      Importer importer = new Importer(importConfig, cfg, envConfig);
      rootContainer = initializeRootContainer(envConfig);
      return importer.processImport(rootContainer);
    }
@@ -1320,36 +1325,41 @@
    boolean openRootContainer = rootContainer == null;

    /*
      If the rootContainer is open, the backend is initialized by something
      else. We can't do any rebuild of system indexes while others are using
      this backend.
   */
     * If the rootContainer is open, the backend is initialized by something
     * else. We can't do any rebuild of system indexes while others are using
     * this backend.
     */
    if(!openRootContainer && rebuildConfig.includesSystemIndex())
    {
      Message message = ERR_JEB_REBUILD_BACKEND_ONLINE.get();
      throw new DirectoryException(DirectoryServer.getServerErrorResultCode(),
              message);
    }
    Importer importer;

    try
    {
      EnvironmentConfig envConfig =
              ConfigurableEnvironment.parseConfigEntry(cfg);
      importer = Importer.getInstance(rebuildConfig, cfg, envConfig);
      EnvironmentConfig envConfig;
      if (openRootContainer)
      {
        envConfig.setReadOnly(false);
        envConfig = new EnvironmentConfig();
        envConfig.setAllowCreate(true);
        envConfig.setTransactional(false);
        envConfig.setConfigParam(EnvironmentConfig.ENV_IS_LOCKING, "true");
        envConfig.setDurability(Durability.COMMIT_NO_SYNC);
        envConfig.setLockTimeout(0, TimeUnit.SECONDS);
        envConfig.setTxnTimeout(0, TimeUnit.SECONDS);
        envConfig.setConfigParam(
                               EnvironmentConfig.ENV_RUN_CHECKPOINTER, "false");
        envConfig.setConfigParam(EnvironmentConfig.ENV_RUN_CLEANER, "false");
        envConfig.setConfigParam(EnvironmentConfig.EVICTOR_LRU_ONLY, "false");
        envConfig.setConfigParam(
                               EnvironmentConfig.EVICTOR_NODES_PER_SCAN, "128");
            EnvironmentConfig.CLEANER_MIN_FILE_UTILIZATION, String.valueOf(cfg
                .getDBCleanerMinUtilization()));
        envConfig.setConfigParam(EnvironmentConfig.LOG_FILE_MAX, String
            .valueOf(cfg.getDBLogFileMax()));
        rootContainer = initializeRootContainer(envConfig);
      }
      else
      {
        envConfig = ConfigurableEnvironment.parseConfigEntry(cfg);
      }

      Importer importer = new Importer(rebuildConfig, cfg, envConfig);
      importer.rebuildIndexes(rootContainer);
    }
    catch (ExecutionException execEx)

 opends/src/server/org/opends/server/backends/jeb/importLDIF/Importer.java

@@ -28,27 +28,35 @@
package org.opends.server.backends.jeb.importLDIF;

import static org.opends.messages.JebMessages.*;
import static org.opends.server.loggers.ErrorLogger.*;
import static org.opends.server.util.DynamicConstants.*;
import static org.opends.server.loggers.ErrorLogger.logError;
import static org.opends.server.util.DynamicConstants.BUILD_ID;
import static org.opends.server.util.DynamicConstants.REVISION_NUMBER;
import static org.opends.server.util.ServerConstants.*;
import static org.opends.server.util.StaticUtils.getFileForPath;

import java.io.*;
import java.nio.*;
import java.nio.ByteBuffer;
import java.nio.channels.FileChannel;
import java.util.*;
import java.util.concurrent.*;
import java.util.concurrent.atomic.*;
import static org.opends.server.util.StaticUtils.getFileForPath;
import org.opends.messages.Message;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicLong;

import org.opends.messages.Category;
import org.opends.messages.Message;
import org.opends.messages.Severity;
import org.opends.server.admin.std.meta.LocalDBIndexCfgDefn;
import org.opends.server.admin.std.meta.LocalDBIndexCfgDefn.IndexType;
import org.opends.server.admin.std.server.LocalDBBackendCfg;
import org.opends.server.admin.std.server.LocalDBIndexCfg;
import org.opends.server.admin.std.meta.LocalDBIndexCfgDefn;
import org.opends.server.backends.jeb.*;
import org.opends.server.config.ConfigException;
import org.opends.server.core.DirectoryServer;
import org.opends.server.types.*;
import org.opends.server.util.*;
import org.opends.server.util.LDIFReader;
import org.opends.server.util.Platform;
import org.opends.server.util.StaticUtils;

import com.sleepycat.je.*;
import com.sleepycat.util.PackedInteger;

@@ -57,10 +65,10 @@
 * This class provides the engine that performs both importing of LDIF files and
 * the rebuilding of indexes.
 */
public class Importer
public final class Importer
{
  private static final int TIMER_INTERVAL = 10000;
  final static int KB = 1024;
  private static final int KB = 1024;
  private static final int MB =  (KB * KB);
  private static final String DEFAULT_TMP_DIR = "import-tmp";
  private static final String TMPENV_DIR = "tmp-env";
@@ -72,31 +80,25 @@

  //Defaults for LDIF reader buffers, min memory required to import and default
  //size for byte buffers.
  private static final int READER_WRITER_BUFFER_SIZE = 1 * MB;
  private static final int READER_WRITER_BUFFER_SIZE = 8 * KB;
  private static final int MIN_DB_CACHE_MEMORY = MAX_DB_CACHE_SIZE +
                                                 MAX_DB_LOG_SIZE;
  private static final int BYTE_BUFFER_CAPACITY = 128;

  //Min and MAX sizes of phase one buffer.
  private static final int MAX_BUFFER_SIZE = 100 * MB;
  private static final int MIN_BUFFER_SIZE = 8 * KB;
  private static final int MAX_BUFFER_SIZE = 2 * MB;
  private static final int MIN_BUFFER_SIZE = 4 * KB;

  //Min size of phase two read-ahead cache.
  private static final int MIN_READ_AHEAD_CACHE_SIZE = 2 * KB;

  //Set aside this much for the JVM from free memory.
  private static final int JVM_MEM_PCT = 45;

  //Percent of import memory to use for temporary environment if the
  //skip DN validation flag isn't specified.
  private static final int TMPENV_MEM_PCT = 50;
  //Small heap threshold used to give more memory to JVM to attempt OOM errors.
  private static final int SMALL_HEAP_SIZE = 256 * MB;

  //The DN attribute type.
  private static AttributeType dnType;
  private static final IndexBuffer.IndexComparator indexComparator =
          new IndexBuffer.IndexComparator();
  static final IndexOutputBuffer.IndexComparator indexComparator =
          new IndexOutputBuffer.IndexComparator();

  //Phase one buffer and imported entries counts.
  private final AtomicInteger bufferCount = new AtomicInteger(0);
@@ -124,15 +126,20 @@
  //Import configuration.
  private final LDIFImportConfig importConfiguration;

  //Backend configuration.
  private final LocalDBBackendCfg backendConfiguration;

  //LDIF reader.
  private LDIFReader reader;

  //Migrated entry count.
  private int migratedCount;

  //Size in bytes of temporary env, DB cache, DB log buf size.
  private long tmpEnvCacheSize = 0, dbCacheSize = MAX_DB_CACHE_SIZE,
               dbLogBufSize = MAX_DB_LOG_SIZE;
  // Size in bytes of temporary env.
  private long tmpEnvCacheSize;

  // Size in bytes of DB cache.
  private long dbCacheSize;

  //The executor service used for the buffer sort tasks.
  private ExecutorService bufferSortService;
@@ -141,14 +148,14 @@
  private ExecutorService scratchFileWriterService;

  //Queue of free index buffers -- used to re-cycle index buffers;
  private final BlockingQueue<IndexBuffer> freeBufferQueue =
          new LinkedBlockingQueue<IndexBuffer>();
  private final BlockingQueue<IndexOutputBuffer> freeBufferQueue =
          new LinkedBlockingQueue<IndexOutputBuffer>();

  //Map of index keys to index buffers.  Used to allocate sorted
  //index buffers to a index writer thread.
  private final
  Map<IndexKey, BlockingQueue<IndexBuffer>> indexKeyQueMap =
          new ConcurrentHashMap<IndexKey, BlockingQueue<IndexBuffer>>();
  Map<IndexKey, BlockingQueue<IndexOutputBuffer>> indexKeyQueMap =
          new ConcurrentHashMap<IndexKey, BlockingQueue<IndexOutputBuffer>>();

  //Map of DB containers to index managers. Used to start phase 2.
  private final List<IndexManager> indexMgrList =
@@ -202,188 +209,158 @@
    }
  }

  //Rebuild-index instance.
  private
  Importer(RebuildConfig rebuildConfig, LocalDBBackendCfg cfg,
            EnvironmentConfig envConfig) throws
          InitializationException, JebException, ConfigException


  /**
   * Create a new import job with the specified rebuild index config.
   *
   * @param rebuildConfig
   *          The rebuild index configuration.
   * @param cfg
   *          The local DB back-end configuration.
   * @param envConfig
   *          The JEB environment config.
   * @throws InitializationException
   *           If a problem occurs during initialization.
   * @throws JebException
   *           If an error occurred when opening the DB.
   * @throws ConfigException
   *           If a problem occurs during initialization.
   */
  public Importer(RebuildConfig rebuildConfig, LocalDBBackendCfg cfg,
      EnvironmentConfig envConfig) throws  InitializationException,
      JebException, ConfigException
  {
    importConfiguration = null;
    tmpEnv = null;
    threadCount = 1;
    rebuildManager = new RebuildIndexManager(rebuildConfig, cfg);
    indexCount = rebuildManager.getIndexCount();
    scratchFileWriterList = new ArrayList<ScratchFileWriterTask>(indexCount);
    scratchFileWriterFutures = new CopyOnWriteArrayList<Future<?>>();
    this.importConfiguration = null;
    this.backendConfiguration = cfg;
    this.tmpEnv = null;
    this.threadCount = 1;
    this.rebuildManager = new RebuildIndexManager(rebuildConfig, cfg);
    this.indexCount = rebuildManager.getIndexCount();
    this.scratchFileWriterList = new ArrayList<ScratchFileWriterTask>(
        indexCount);
    this.scratchFileWriterFutures = new CopyOnWriteArrayList<Future<?>>();

    File parentDir;
    if(rebuildConfig.getTmpDirectory() == null)
    if (rebuildConfig.getTmpDirectory() == null)
    {
      parentDir = getFileForPath(DEFAULT_TMP_DIR);
    }
    else
    {
       parentDir = getFileForPath(rebuildConfig.getTmpDirectory());
      parentDir = getFileForPath(rebuildConfig.getTmpDirectory());
    }
    tempDir = new File(parentDir, cfg.getBackendId());

    this.tempDir = new File(parentDir, cfg.getBackendId());
    recursiveDelete(tempDir);
    if(!tempDir.exists() && !tempDir.mkdirs())
    if (!tempDir.exists() && !tempDir.mkdirs())
    {
      Message message =
                ERR_JEB_IMPORT_CREATE_TMPDIR_ERROR.get(String.valueOf(tempDir));
      Message message = ERR_JEB_IMPORT_CREATE_TMPDIR_ERROR.get(String
          .valueOf(tempDir));
      throw new InitializationException(message);
    }
    skipDNValidation = true;
    if(envConfig != null)
    {
      initializeDBEnv(envConfig);
    }
    this.skipDNValidation = true;
    initializeDBEnv(envConfig);
  }



  /**
   * Create a new import job with the specified ldif import config.
   *
   * @param importConfiguration The LDIF import configuration.
   * @param localDBBackendCfg The local DB back-end configuration.
   * @param envConfig The JEB environment config.
   * @throws  InitializationException If a problem occurs during initialization.
   * @param importConfiguration
   *          The LDIF import configuration.
   * @param localDBBackendCfg
   *          The local DB back-end configuration.
   * @param envConfig
   *          The JEB environment config.
   * @throws InitializationException
   *           If a problem occurs during initialization.
   * @throws ConfigException
   *           If a problem occurs reading the configuration.
   * @throws DatabaseException
   *           If an error occurred when opening the DB.
   */
  private Importer(LDIFImportConfig importConfiguration,
                   LocalDBBackendCfg localDBBackendCfg,
                   EnvironmentConfig envConfig) throws
          InitializationException, DatabaseException
  public Importer(LDIFImportConfig importConfiguration,
      LocalDBBackendCfg localDBBackendCfg, EnvironmentConfig envConfig)
      throws InitializationException, ConfigException, DatabaseException
  {
    rebuildManager = null;
    this.rebuildManager = null;
    this.importConfiguration = importConfiguration;
    if(importConfiguration.getThreadCount() == 0)
    this.backendConfiguration = localDBBackendCfg;

    if (importConfiguration.getThreadCount() == 0)
    {
      threadCount = Runtime.getRuntime().availableProcessors() * 2;
      this.threadCount = Runtime.getRuntime().availableProcessors() * 2;
    }
    else
    {
      threadCount = importConfiguration.getThreadCount();
      this.threadCount = importConfiguration.getThreadCount();
    }
    indexCount = localDBBackendCfg.listLocalDBIndexes().length + 2;
    if(!importConfiguration.appendToExistingData()) {
      if(importConfiguration.clearBackend() ||
              localDBBackendCfg.getBaseDN().size() <= 1) {
        clearedBackend = true;

    // Determine the number of indexes.
    int indexes = 2; // dn2id + dn2uri
    for (String indexName : localDBBackendCfg.listLocalDBIndexes())
    {
      LocalDBIndexCfg index = localDBBackendCfg.getLocalDBIndex(indexName);
      SortedSet<IndexType> types = index.getIndexType();
      if (types.contains(IndexType.EXTENSIBLE))
      {
        indexes += types.size() - 1
            + index.getIndexExtensibleMatchingRule().size();
      }
      else
      {
        indexes += types.size();
      }
    }
    scratchFileWriterList = new ArrayList<ScratchFileWriterTask>(indexCount);
    scratchFileWriterFutures = new CopyOnWriteArrayList<Future<?>>();
    this.indexCount = indexes;

    if (!importConfiguration.appendToExistingData())
    {
      if (importConfiguration.clearBackend()
          || localDBBackendCfg.getBaseDN().size() <= 1)
      {
        this.clearedBackend = true;
      }
    }
    this.scratchFileWriterList = new ArrayList<ScratchFileWriterTask>(
        indexCount);
    this.scratchFileWriterFutures = new CopyOnWriteArrayList<Future<?>>();
    File parentDir;
    if(importConfiguration.getTmpDirectory() == null)
    if (importConfiguration.getTmpDirectory() == null)
    {
      parentDir = getFileForPath(DEFAULT_TMP_DIR);
    }
    else
    {
       parentDir = getFileForPath(importConfiguration.getTmpDirectory());
      parentDir = getFileForPath(importConfiguration.getTmpDirectory());
    }
    tempDir = new File(parentDir, localDBBackendCfg.getBackendId());
    this.tempDir = new File(parentDir, localDBBackendCfg.getBackendId());
    recursiveDelete(tempDir);
    if(!tempDir.exists() && !tempDir.mkdirs())
    if (!tempDir.exists() && !tempDir.mkdirs())
    {
      Message message =
                ERR_JEB_IMPORT_CREATE_TMPDIR_ERROR.get(String.valueOf(tempDir));
      Message message = ERR_JEB_IMPORT_CREATE_TMPDIR_ERROR.get(String
          .valueOf(tempDir));
      throw new InitializationException(message);
    }
    skipDNValidation = importConfiguration.getSkipDNValidation();
    initializeDBEnv(envConfig);
    //Set up temporary environment.
    if(!skipDNValidation)

    // Set up temporary environment.
    if (!skipDNValidation)
    {
      File envPath = new File(tempDir, TMPENV_DIR);
      envPath.mkdirs();
      tmpEnv = new TmpEnv(envPath);
      this.tmpEnv = new TmpEnv(envPath);
    }
    else
    {
      tmpEnv = null;
      this.tmpEnv = null;
    }
  }


  /**
   * Return and import LDIF instance using the specified arguments.
   *
   * @param importCfg The import config to use.
   * @param localDBBackendCfg The local DB backend config to use.
   * @param envCfg The JEB environment config to use.
   * @return A import LDIF instance.
   *
   * @throws InitializationException If the instance cannot be initialized.
   */
  public static
  Importer getInstance(LDIFImportConfig importCfg,
                       LocalDBBackendCfg localDBBackendCfg,
                       EnvironmentConfig envCfg)
          throws InitializationException
  {
     return  new Importer(importCfg, localDBBackendCfg, envCfg);
  }


  /**
   * Return an import rebuild index instance using the specified arguments.
   *
   * @param rebuildCfg The rebuild config to use.
   * @param localDBBackendCfg The local DB backend config to use.
   * @param envCfg The JEB environment config to use.
   * @return An import rebuild index instance.
   *
   * @throws InitializationException If the instance cannot be initialized.
   * @throws JebException If a JEB exception occurs.
   * @throws ConfigException If the instance cannot be configured.
   */
  public static synchronized
  Importer getInstance(RebuildConfig rebuildCfg,
                       LocalDBBackendCfg localDBBackendCfg,
                       EnvironmentConfig envCfg)
  throws InitializationException, JebException, ConfigException
  {
      return new Importer(rebuildCfg, localDBBackendCfg, envCfg);
  }


  private void adjustBufferSize(long availMem)
  {
     int oldThreadCount = threadCount;
     for(;threadCount > 0; threadCount--)
     {
       phaseOneBufferCount = 2 * (indexCount * threadCount);
       bufferSize = (int) (availMem/ ((4 * indexCount) + phaseOneBufferCount));
       if(bufferSize >= MIN_BUFFER_SIZE)
       {
         break;
       }
     }
     Message message =
           NOTE_JEB_IMPORT_ADJUST_THREAD_COUNT.get(oldThreadCount, threadCount);
     logError(message);
  }


  private boolean getBufferSizes(long availMem)
  {
    boolean maxBuf = false;
    bufferSize = (int) (availMem / ((4 * indexCount) + phaseOneBufferCount));
    if(bufferSize >= MIN_BUFFER_SIZE)
    {
      if(bufferSize > MAX_BUFFER_SIZE)
      {
        bufferSize = MAX_BUFFER_SIZE;
        maxBuf = true;
      }
    }
    else if(bufferSize < MIN_BUFFER_SIZE)
    {
      adjustBufferSize(availMem);
    }
    return maxBuf;
  }


  /**
   * Return the suffix instance in the specified map that matches the specified
@@ -410,155 +387,204 @@
  }


  private long getTmpEnvironmentMemory(long availableMemoryImport)
  {
    int tmpMemPct = TMPENV_MEM_PCT;
    tmpEnvCacheSize = (availableMemoryImport * tmpMemPct) / 100;
    availableMemoryImport -= tmpEnvCacheSize;
    if(!clearedBackend)
    {
      long additionalDBCache = (tmpEnvCacheSize * 85) / 100;
      tmpEnvCacheSize -= additionalDBCache;
      dbCacheSize += additionalDBCache;
    }
    return availableMemoryImport;
  }


  //Used for large heap sizes when the buffer max size has been identified. Any
  //extra memory can be given to the temporary environment in that case.
  private void adjustTmpEnvironmentMemory(long availableMemoryImport)
  {
    long additionalMem = availableMemoryImport -
            (phaseOneBufferCount * MAX_BUFFER_SIZE);
    if(additionalMem > 0)
    {
      tmpEnvCacheSize += additionalMem;
      if(!clearedBackend)
      {
        //The DN cache probably needs to be smaller and the DB cache bigger
        //because the dn2id is checked if the backend has not been cleared.
        long additionalDBCache = (tmpEnvCacheSize * 85) / 100;
        tmpEnvCacheSize -= additionalDBCache;
        dbCacheSize += additionalDBCache;
      }
    }
  }


  private long defaultMemoryCalc(long availMem)
          throws InitializationException
  {
    long bufMem;
    if(availMem < (MIN_DB_CACHE_MEMORY + MIN_DB_CACHE_SIZE))
    {
      long minCacheSize = MIN_DB_CACHE_SIZE;
      if (System.getProperty(PROPERTY_RUNNING_UNIT_TESTS) != null) {
        minCacheSize = 500 *KB;
      }
      dbCacheSize = minCacheSize;
      tmpEnvCacheSize = minCacheSize;
      dbLogBufSize = 0;
      bufMem = availMem - 2 * minCacheSize;
      if(bufMem < 0 || (bufMem < (2 * indexCount) * MIN_BUFFER_SIZE)) {
        Message message =
              ERR_IMPORT_LDIF_LACK_MEM.get(availMem,
                 ((2 * indexCount) * MIN_BUFFER_SIZE) + 2 * MIN_DB_CACHE_SIZE);
        throw new InitializationException(message);
      }
    }
    else
    {
      bufMem = getTmpEnvironmentMemory(availMem);
    }
    return bufMem;
  }

  private long skipDNValidationCalc(long availMem)
          throws InitializationException
  {
    long bufMem = availMem;
    if(availMem < (MIN_DB_CACHE_MEMORY))
    {
      long minCacheSize = MIN_DB_CACHE_SIZE;
      if (System.getProperty(PROPERTY_RUNNING_UNIT_TESTS) != null) {
        minCacheSize = 500 *KB;
      }
      dbCacheSize = minCacheSize;
      dbLogBufSize = 0;
      bufMem = availMem - minCacheSize;
      if(bufMem < 0 || (bufMem < (2 * indexCount) * MIN_BUFFER_SIZE)) {
        Message message =
              ERR_IMPORT_LDIF_LACK_MEM.get(availMem,
                  ((2 * indexCount) * MIN_BUFFER_SIZE) + MIN_DB_CACHE_SIZE);
        throw new InitializationException(message);
      }
    }
    return bufMem;
  }

  /**
   * Calculate buffer sizes and initialize JEB properties based on memory.
   *
   * @param envConfig The environment config to use in the calculations.
   *
   * @throws InitializationException If a problem occurs during calculation.
   * @param envConfig
   *          The environment config to use in the calculations.
   * @throws InitializationException
   *           If a problem occurs during calculation.
   */
  private void initializeDBEnv(EnvironmentConfig envConfig)
          throws InitializationException
      throws InitializationException
  {
      Message message;
      phaseOneBufferCount = 2 * (indexCount * threadCount);
      Runtime runTime = Runtime.getRuntime();
      long totFreeMemory = runTime.freeMemory() +
                            (runTime.maxMemory() - runTime.totalMemory());
      int importMemPct = (100 - JVM_MEM_PCT);
      if(totFreeMemory <= SMALL_HEAP_SIZE)
    // Calculate amount of usable memory. This will need to take into account
    // various fudge factors, including the number of IO buffers used by the
    // scratch writers (1 per index).
    final long availableMemory = calculateAvailableMemory();
    final long usableMemory = availableMemory
        - (indexCount * READER_WRITER_BUFFER_SIZE);

    if (!skipDNValidation)
    {
      // No DN validation: calculate memory for DB cache, DN2ID temporary cache,
      // and buffers.
      if (System.getProperty(PROPERTY_RUNNING_UNIT_TESTS) != null)
      {
        importMemPct -= 15;
        dbCacheSize = 500 * KB;
        tmpEnvCacheSize = 500 * KB;
      }
      if(rebuildManager != null)
      else if (usableMemory < (MIN_DB_CACHE_MEMORY + MIN_DB_CACHE_SIZE))
      {
        importMemPct -= 15;
        dbCacheSize = MIN_DB_CACHE_SIZE;
        tmpEnvCacheSize = MIN_DB_CACHE_SIZE;
      }
      long phaseOneBufferMemory;
      if(!skipDNValidation)
      else if (!clearedBackend)
      {
        phaseOneBufferMemory =
                       defaultMemoryCalc((totFreeMemory * importMemPct) / 100);
        // Appending to existing data so reserve extra memory for the DB cache
        // since it will be needed for dn2id queries.
        dbCacheSize = usableMemory * 33 / 100;
        tmpEnvCacheSize = usableMemory * 33 / 100;
      }
      else
      {
        phaseOneBufferMemory =
                     skipDNValidationCalc((totFreeMemory * importMemPct) / 100);
        dbCacheSize = MAX_DB_CACHE_SIZE;
        tmpEnvCacheSize = usableMemory * 66 / 100;
      }
      boolean maxBuffers = getBufferSizes(phaseOneBufferMemory);
      //Give any extra memory to the temp environment cache if there is any.
      if(!skipDNValidation && maxBuffers)
    }
    else
    {
      // No DN validation: calculate memory for DB cache and buffers.

      // No need for DN2ID cache.
      tmpEnvCacheSize = 0;

      if (System.getProperty(PROPERTY_RUNNING_UNIT_TESTS) != null)
      {
        adjustTmpEnvironmentMemory(phaseOneBufferMemory);
        dbCacheSize = 500 * KB;
      }
      message = NOTE_JEB_IMPORT_LDIF_TOT_MEM_BUF.get(phaseOneBufferMemory,
              phaseOneBufferCount);
      else if (usableMemory < MIN_DB_CACHE_MEMORY)
      {
        dbCacheSize = MIN_DB_CACHE_SIZE;
      }
      else
      {
        // No need to differentiate between append/clear backend, since dn2id is
        // not being queried.
        dbCacheSize = MAX_DB_CACHE_SIZE;
      }
    }

    final long phaseOneBufferMemory = usableMemory - dbCacheSize
        - tmpEnvCacheSize;
    final int oldThreadCount = threadCount;
    while (true)
    {
      phaseOneBufferCount = 2 * indexCount * threadCount;

      // Scratch writers allocate 4 buffers per index as well.
      final int totalPhaseOneBufferCount = phaseOneBufferCount
          + (4 * indexCount);
      bufferSize = (int) (phaseOneBufferMemory / totalPhaseOneBufferCount);

      if (bufferSize > MAX_BUFFER_SIZE)
      {
        if (!skipDNValidation)
        {
          // The buffers are big enough: the memory is best used for the DN2ID
          // temp DB.
          bufferSize = MAX_BUFFER_SIZE;

          final long extraMemory = phaseOneBufferMemory
              - (totalPhaseOneBufferCount * bufferSize);
          if (!clearedBackend)
          {
            dbCacheSize += extraMemory / 2;
            tmpEnvCacheSize += extraMemory / 2;
          }
          else
          {
            tmpEnvCacheSize += extraMemory;
          }
        }

        break;
      }
      else if (bufferSize > MIN_BUFFER_SIZE)
      {
        // This is acceptable.
        break;
      }
      else if (threadCount > 1)
      {
        // Retry using less threads.
        threadCount--;
      }
      else
      {
        // Not enough memory.
        final long minimumPhaseOneBufferMemory = totalPhaseOneBufferCount
            * MIN_BUFFER_SIZE;
        Message message = ERR_IMPORT_LDIF_LACK_MEM.get(usableMemory,
            minimumPhaseOneBufferMemory + dbCacheSize + tmpEnvCacheSize);
        throw new InitializationException(message);
      }
    }

    if (oldThreadCount != threadCount)
    {
      Message message = NOTE_JEB_IMPORT_ADJUST_THREAD_COUNT.get(oldThreadCount,
          threadCount);
      logError(message);
      if(tmpEnvCacheSize > 0)
      {
         message = NOTE_JEB_IMPORT_LDIF_TMP_ENV_MEM.get(tmpEnvCacheSize);
        logError(message);
      }
      envConfig.setConfigParam(EnvironmentConfig.ENV_RUN_CLEANER, "true");
      envConfig.setConfigParam(EnvironmentConfig.MAX_MEMORY,
              Long.toString(dbCacheSize));
      message = NOTE_JEB_IMPORT_LDIF_DB_MEM_BUF_INFO.get(dbCacheSize,
              bufferSize);
    }

    Message message = NOTE_JEB_IMPORT_LDIF_TOT_MEM_BUF.get(
        phaseOneBufferMemory, phaseOneBufferCount);
    logError(message);
    if (tmpEnvCacheSize > 0)
    {
      message = NOTE_JEB_IMPORT_LDIF_TMP_ENV_MEM.get(tmpEnvCacheSize);
      logError(message);
      if(dbLogBufSize > 0)
    }
    envConfig.setConfigParam(EnvironmentConfig.MAX_MEMORY, Long
        .toString(dbCacheSize));
    message = NOTE_JEB_IMPORT_LDIF_DB_MEM_BUF_INFO.get(dbCacheSize, bufferSize);
    logError(message);
  }



  /**
   * Returns the amount of available memory which can be used by this import,
   * taking into account whether or not the import is running offline or online
   * as a task.
   */
  private long calculateAvailableMemory()
  {
    final long availableMemory;
    if (DirectoryServer.isRunning())
    {
      // Online import/rebuild.
      Runtime runTime = Runtime.getRuntime();
      runTime.gc();
      runTime.gc();
      final long usedMemory = runTime.totalMemory() - runTime.freeMemory();
      final long maxUsableMemory = Platform.getUsableMemoryForCaching();
      final long usableMemory = maxUsableMemory - usedMemory;

      final long configuredMemory;
      if (backendConfiguration.getDBCacheSize() > 0)
      {
        envConfig.setConfigParam(EnvironmentConfig.LOG_TOTAL_BUFFER_BYTES,
                Long.toString(MAX_DB_LOG_SIZE));
        message = NOTE_JEB_IMPORT_LDIF_LOG_BYTES.get(MAX_DB_LOG_SIZE);
        logError(message);
        configuredMemory = backendConfiguration.getDBCacheSize();
      }
      else
      {
        configuredMemory = backendConfiguration.getDBCachePercent()
            * Runtime.getRuntime().maxMemory() / 100;
      }
      availableMemory = Math.min(usableMemory, configuredMemory);
    }
    else
    {
      // Offline import/rebuild.
      availableMemory = Platform.getUsableMemoryForCaching();
    }

    // Now take into account various fudge factors.
    int importMemPct = 90;
    if (availableMemory <= SMALL_HEAP_SIZE)
    {
      // Be pessimistic when memory is low.
      importMemPct -= 25;
    }
    if (rebuildManager != null)
    {
      // Rebuild seems to require more overhead.
      importMemPct -= 15;
    }

    return (availableMemory * importMemPct / 100);
  }


@@ -566,7 +592,7 @@
  {
    for(int i = 0; i < phaseOneBufferCount; i++)
    {
      IndexBuffer b = IndexBuffer.createIndexBuffer(bufferSize);
      IndexOutputBuffer b = new IndexOutputBuffer(bufferSize);
      freeBufferQueue.add(b);
    }
  }
@@ -802,10 +828,10 @@
    this.rootContainer = rootContainer;
    try
    {
    reader = new LDIFReader(importConfiguration, rootContainer,
                                 READER_WRITER_BUFFER_SIZE);
      reader = new LDIFReader(importConfiguration, rootContainer,
          READER_WRITER_BUFFER_SIZE);
    }
    catch(IOException ioe)
    catch (IOException ioe)
    {
      Message message = ERR_JEB_IMPORT_LDIF_READER_IO_ERROR.get();
      throw new InitializationException(message, ioe);
@@ -813,9 +839,8 @@

    try
    {
      Message message =
              NOTE_JEB_IMPORT_STARTING.get(DirectoryServer.getVersionString(),
                      BUILD_ID, REVISION_NUMBER);
      Message message = NOTE_JEB_IMPORT_STARTING.get(
          DirectoryServer.getVersionString(), BUILD_ID, REVISION_NUMBER);
      logError(message);
      message = NOTE_JEB_IMPORT_THREAD_COUNT.get(threadCount);
      logError(message);
@@ -823,11 +848,11 @@
      long startTime = System.currentTimeMillis();
      phaseOne();
      long phaseOneFinishTime = System.currentTimeMillis();
      if(!skipDNValidation)
      if (!skipDNValidation)
      {
         tmpEnv.shutdown();
        tmpEnv.shutdown();
      }
      if(isPhaseOneCanceled)
      if (isPhaseOneCanceled)
      {
        throw new InterruptedException("Import processing canceled.");
      }
@@ -840,24 +865,22 @@
      long finishTime = System.currentTimeMillis();
      long importTime = (finishTime - startTime);
      float rate = 0;
      message = NOTE_JEB_IMPORT_PHASE_STATS.get(importTime/1000,
                        (phaseOneFinishTime - startTime)/1000,
                        (phaseTwoFinishTime - phaseTwoTime)/1000);
      message = NOTE_JEB_IMPORT_PHASE_STATS.get(importTime / 1000,
          (phaseOneFinishTime - startTime) / 1000,
          (phaseTwoFinishTime - phaseTwoTime) / 1000);
      logError(message);
      if (importTime > 0)
        rate = 1000f * reader.getEntriesRead() / importTime;
        message = NOTE_JEB_IMPORT_FINAL_STATUS.get(reader.getEntriesRead(),
                  importCount.get(), reader.getEntriesIgnored(),
                  reader.getEntriesRejected(), migratedCount,
                  importTime / 1000, rate);
      if (importTime > 0) rate = 1000f * reader.getEntriesRead() / importTime;
      message = NOTE_JEB_IMPORT_FINAL_STATUS.get(reader.getEntriesRead(),
          importCount.get(), reader.getEntriesIgnored(),
          reader.getEntriesRejected(), migratedCount, importTime / 1000, rate);
      logError(message);
    }
    finally
    {
      reader.close();
    }
    return new LDIFImportResult(reader.getEntriesRead(), reader
            .getEntriesRejected(), reader.getEntriesIgnored());
    return new LDIFImportResult(reader.getEntriesRead(),
        reader.getEntriesRejected(), reader.getEntriesIgnored());
  }


@@ -921,31 +944,34 @@
  }



  private void phaseOne() throws InterruptedException, ExecutionException
  {
    initializeIndexBuffers();
    FirstPhaseProgressTask progressTask = new FirstPhaseProgressTask();
    ScheduledThreadPoolExecutor timerService =
      new ScheduledThreadPoolExecutor(1);
    ScheduledThreadPoolExecutor timerService = new ScheduledThreadPoolExecutor(
        1);
    timerService.scheduleAtFixedRate(progressTask, TIMER_INTERVAL,
      TIMER_INTERVAL, TimeUnit.MILLISECONDS);
        TIMER_INTERVAL, TimeUnit.MILLISECONDS);
    scratchFileWriterService = Executors.newFixedThreadPool(2 * indexCount);
    bufferSortService = Executors.newFixedThreadPool(threadCount);
    ExecutorService execService = Executors.newFixedThreadPool(threadCount);
    List<Callable<Void>> tasks = new ArrayList<Callable<Void>>(threadCount);
    tasks.add(new MigrateExistingTask());
    List<Future<Void>> results = execService.invokeAll(tasks);
    for (Future<Void> result : results) {
      if(!result.isDone()) {
    for (Future<Void> result : results)
    {
      if (!result.isDone())
      {
        result.get();
      }
    }
    tasks.clear();
    results.clear();
    if (importConfiguration.appendToExistingData() &&
            importConfiguration.replaceExistingEntries())
    if (importConfiguration.appendToExistingData()
        && importConfiguration.replaceExistingEntries())
    {
     for (int i = 0; i < threadCount; i++)
      for (int i = 0; i < threadCount; i++)
      {
        tasks.add(new AppendReplaceTask());
      }
@@ -959,24 +985,32 @@
    }
    results = execService.invokeAll(tasks);
    for (Future<Void> result : results)
      if(!result.isDone()) {
    {
      if (!result.isDone())
      {
        result.get();
      }
    }
    tasks.clear();
    results.clear();
    tasks.add(new MigrateExcludedTask());
    results = execService.invokeAll(tasks);
    for (Future<Void> result : results)
      if(!result.isDone()) {
        result.get();
      }
    stopScratchFileWriters();
    for (Future<?> result : scratchFileWriterFutures)
    {
     if(!result.isDone()) {
      if (!result.isDone())
      {
        result.get();
      }
    }
    stopScratchFileWriters();
    for (Future<?> result : scratchFileWriterFutures)
    {
      if (!result.isDone())
      {
        result.get();
      }
    }

    // Shutdown the executor services
    timerService.shutdown();
    timerService.awaitTermination(30, TimeUnit.SECONDS);
@@ -987,7 +1021,7 @@
    scratchFileWriterService.shutdown();
    scratchFileWriterService.awaitTermination(30, TimeUnit.SECONDS);

    //Try to clear as much memory as possible.
    // Try to clear as much memory as possible.
    scratchFileWriterList.clear();
    scratchFileWriterFutures.clear();
    indexKeyQueMap.clear();
@@ -996,41 +1030,31 @@



  private void phaseTwo() throws InterruptedException, JebException,
          ExecutionException
  private void phaseTwo() throws InitializationException, InterruptedException,
      JebException, ExecutionException
  {
    SecondPhaseProgressTask progress2Task =
            new SecondPhaseProgressTask(reader.getEntriesRead());
    ScheduledThreadPoolExecutor timerService =
      new ScheduledThreadPoolExecutor(1);
    SecondPhaseProgressTask progress2Task = new SecondPhaseProgressTask(
        reader.getEntriesRead());
    ScheduledThreadPoolExecutor timerService = new ScheduledThreadPoolExecutor(
        1);
    timerService.scheduleAtFixedRate(progress2Task, TIMER_INTERVAL,
      TIMER_INTERVAL, TimeUnit.MILLISECONDS);
    processIndexFiles();
    timerService.shutdown();
    timerService.awaitTermination(30, TimeUnit.SECONDS);
  }


  private int getBufferCount(int dbThreads)
  {
    int buffers = 0;

    List<IndexManager> totList = new LinkedList<IndexManager>(DNIndexMgrList);
    totList.addAll(indexMgrList);
    Collections.sort(totList, Collections.reverseOrder());
    int limit = Math.min(dbThreads, totList.size());
    for(int i = 0; i < limit; i ++)
        TIMER_INTERVAL, TimeUnit.MILLISECONDS);
    try
    {
      buffers += totList.get(i).getBufferList().size();
      processIndexFiles();
    }
    return buffers;
    finally
    {
      timerService.shutdown();
      timerService.awaitTermination(30, TimeUnit.SECONDS);
    }
  }


  private void processIndexFiles() throws InterruptedException,
          JebException, ExecutionException

  private void processIndexFiles() throws InitializationException,
      InterruptedException, JebException, ExecutionException
  {
    ExecutorService dbService;
    if(bufferCount.get() == 0)
    {
      return;
@@ -1040,60 +1064,89 @@
    {
      dbThreads = 4;
    }
    int readAheadSize =  cacheSizeFromFreeMemory(getBufferCount(dbThreads));

    // Calculate memory / buffer counts.
    final long availableMemory = calculateAvailableMemory();
    final long usableMemory = availableMemory - dbCacheSize;

    int readAheadSize;
    int buffers;
    while (true)
    {
      final List<IndexManager> totList = new ArrayList<IndexManager>(
          DNIndexMgrList);
      totList.addAll(indexMgrList);
      Collections.sort(totList, Collections.reverseOrder());

      buffers = 0;
      final int limit = Math.min(dbThreads, totList.size());
      for (int i = 0; i < limit; i++)
      {
        buffers += totList.get(i).bufferIndexCount;
      }

      readAheadSize = (int) (usableMemory / buffers);
      if (readAheadSize > bufferSize)
      {
        // Cache size is never larger than the buffer size.
        readAheadSize = bufferSize;
        break;
      }
      else if (readAheadSize > MIN_READ_AHEAD_CACHE_SIZE)
      {
        // This is acceptable.
        break;
      }
      else if (dbThreads > 1)
      {
        // Reduce thread count.
        dbThreads--;
      }
      else
      {
        // Not enough memory.
        final long minimumPhaseTwoBufferMemory = buffers
            * MIN_READ_AHEAD_CACHE_SIZE;
        Message message = ERR_IMPORT_LDIF_LACK_MEM.get(usableMemory,
            minimumPhaseTwoBufferMemory + dbCacheSize);
        throw new InitializationException(message);
      }
    }

    Message message = NOTE_JEB_IMPORT_LDIF_PHASE_TWO_MEM_REPORT.get(
        availableMemory, readAheadSize, buffers);
    logError(message);

    // Start indexing tasks.
    List<Future<Void>> futures = new LinkedList<Future<Void>>();
    dbService = Executors.newFixedThreadPool(dbThreads);
    //Start DN processing first.
    for(IndexManager dnMgr : DNIndexMgrList)
    ExecutorService dbService = Executors.newFixedThreadPool(dbThreads);

    // Start DN processing first.
    for (IndexManager dnMgr : DNIndexMgrList)
    {
      futures.add(dbService.submit(new IndexDBWriteTask(dnMgr, readAheadSize)));
    }
    for(IndexManager mgr : indexMgrList)
    for (IndexManager mgr : indexMgrList)
    {
       futures.add(dbService.submit(new IndexDBWriteTask(mgr, readAheadSize)));
      futures.add(dbService.submit(new IndexDBWriteTask(mgr, readAheadSize)));
    }

    for (Future<Void> result : futures)
      if(!result.isDone()) {
    {
      if (!result.isDone())
      {
        result.get();
      }
    }

    dbService.shutdown();
  }


  private int cacheSizeFromFreeMemory(int buffers)
  {
    Runtime runTime = Runtime.getRuntime();
    runTime.gc();
    runTime.gc();
    long freeMemory = runTime.freeMemory();
    long maxMemory = runTime.maxMemory();
    long totMemory = runTime.totalMemory();
    long totFreeMemory = (freeMemory + (maxMemory - totMemory));
    int importMemPct = (100 - JVM_MEM_PCT);
    //For very small heaps, give more memory to the JVM.
    if(totFreeMemory <= SMALL_HEAP_SIZE)
    {
        importMemPct -= 35;
    }
    long availableMemory = (totFreeMemory * importMemPct) / 100;
    int averageBufferSize = (int)(availableMemory /buffers);
    int cacheSize = Math.max(MIN_READ_AHEAD_CACHE_SIZE, averageBufferSize);
    //Cache size is never larger than the buffer size.
    if(cacheSize > bufferSize)
    {
      cacheSize = bufferSize;
    }
    Message message =
     NOTE_JEB_IMPORT_LDIF_PHASE_TWO_MEM_REPORT.get(availableMemory,
                                                   cacheSize, buffers);
    logError(message);
    return cacheSize;
  }


  private void stopScratchFileWriters()
  {
    IndexBuffer indexBuffer = IndexBuffer.createIndexBuffer(0);
    IndexOutputBuffer indexBuffer = new IndexOutputBuffer(0);
    for(ScratchFileWriterTask task : scratchFileWriterList)
    {
      task.queue.add(indexBuffer);
@@ -1293,7 +1346,7 @@
        {
          if (importConfiguration.isCancelled() || isPhaseOneCanceled)
          {
            IndexBuffer indexBuffer = IndexBuffer.createIndexBuffer(0);
            IndexOutputBuffer indexBuffer = new IndexOutputBuffer(0);
            freeBufferQueue.add(indexBuffer);
            return null;
          }
@@ -1462,8 +1515,8 @@
   */
  private  class ImportTask implements Callable<Void>
  {
    private final Map<IndexKey, IndexBuffer> indexBufferMap =
                                     new HashMap<IndexKey, IndexBuffer>();
    private final Map<IndexKey, IndexOutputBuffer> indexBufferMap =
                                     new HashMap<IndexKey, IndexOutputBuffer>();
    private final Set<byte[]> insertKeySet = new HashSet<byte[]>();
    private final EntryInformation entryInfo = new EntryInformation();
    private DatabaseEntry keyEntry = new DatabaseEntry(),
@@ -1481,7 +1534,7 @@
        {
          if (importConfiguration.isCancelled() || isPhaseOneCanceled)
          {
            IndexBuffer indexBuffer = IndexBuffer.createIndexBuffer(0);
            IndexOutputBuffer indexBuffer = new IndexOutputBuffer(0);
            freeBufferQueue.add(indexBuffer);
            return null;
          }
@@ -1653,15 +1706,16 @@
    void flushIndexBuffers() throws InterruptedException,
                 ExecutionException
    {
       Set<Map.Entry<IndexKey, IndexBuffer>> set = indexBufferMap.entrySet();
       Iterator<Map.Entry<IndexKey, IndexBuffer>> setIterator = set.iterator();
       Set<Map.Entry<IndexKey, IndexOutputBuffer>> set =
         indexBufferMap.entrySet();
       Iterator<Map.Entry<IndexKey, IndexOutputBuffer>> setIterator =
         set.iterator();
       while(setIterator.hasNext())
        {
          Map.Entry<IndexKey, IndexBuffer> e = setIterator.next();
          Map.Entry<IndexKey, IndexOutputBuffer> e = setIterator.next();
          IndexKey indexKey = e.getKey();
          IndexBuffer indexBuffer = e.getValue();
          IndexOutputBuffer indexBuffer = e.getValue();
          setIterator.remove();
          ImportIndexType indexType = indexKey.getIndexType();
          indexBuffer.setComparator(indexComparator);
          indexBuffer.setIndexKey(indexKey);
          indexBuffer.setDiscard();
@@ -1674,27 +1728,22 @@

    int
    processKey(DatabaseContainer container, byte[] key, EntryID entryID,
         IndexBuffer.ComparatorBuffer<byte[]> comparator, IndexKey indexKey,
         boolean insert)
         IndexOutputBuffer.ComparatorBuffer<byte[]> comparator,
         IndexKey indexKey, boolean insert)
         throws ConfigException, InterruptedException
    {
      IndexBuffer indexBuffer;
      if(!indexBufferMap.containsKey(indexKey))
      IndexOutputBuffer indexBuffer = indexBufferMap.get(indexKey);
      if (indexBuffer == null)
      {
        indexBuffer = getNewIndexBuffer();
        indexBufferMap.put(indexKey, indexBuffer);
      }
      else
      {
        indexBuffer = indexBufferMap.get(indexKey);
      }
      if(!indexBuffer.isSpaceAvailable(key, entryID.longValue()))
      else if (!indexBuffer.isSpaceAvailable(key, entryID.longValue()))
      {
        indexBuffer.setComparator(comparator);
        indexBuffer.setIndexKey(indexKey);
        bufferSortService.submit(new SortTask(indexBuffer));
        indexBuffer = getNewIndexBuffer();
        indexBufferMap.remove(indexKey);
        indexBufferMap.put(indexKey, indexBuffer);
      }
      int id = System.identityHashCode(container);
@@ -1703,9 +1752,10 @@
    }


    IndexBuffer getNewIndexBuffer() throws ConfigException, InterruptedException
    IndexOutputBuffer getNewIndexBuffer() throws ConfigException,
      InterruptedException
    {
      IndexBuffer indexBuffer = freeBufferQueue.take();
      IndexOutputBuffer indexBuffer = freeBufferQueue.take();
        if(indexBuffer == null)
        {
         Message message = Message.raw(Category.JEB, Severity.SEVERE_ERROR,
@@ -1774,15 +1824,25 @@
    }


    private SortedSet<Buffer> initializeBuffers() throws IOException
    private NavigableSet<IndexInputBuffer> initializeBuffers()
      throws IOException
    {
      SortedSet<Buffer> bufferSet = new TreeSet<Buffer>();
      for(Buffer b : indexMgr.getBufferList())
      NavigableSet<IndexInputBuffer> bufferSet =
        new TreeSet<IndexInputBuffer>();
      for (int i = 0; i < indexMgr.bufferIndexCount; i++)
      {
        b.initializeCache(indexMgr, null, cacheSize);
        IndexInputBuffer b = new IndexInputBuffer(indexMgr,
            indexMgr.bufferIndexBegin[i], indexMgr.bufferIndexEnd[i],
            indexMgr.bufferIndexID[i]);
        b.initializeCache(cacheSize);
        bufferSet.add(b);
      }
      indexMgr.getBufferList().clear();

      // GC arrays.
      indexMgr.bufferIndexBegin = null;
      indexMgr.bufferIndexEnd = null;
      indexMgr.bufferIndexID = null;

      return bufferSet;
    }

@@ -1792,78 +1852,103 @@
     */
    public Void call() throws Exception
    {
      ByteBuffer cKey = null;
      ImportIDSet cInsertIDSet =  new ImportIDSet(),
                  cDeleteIDSet =  new ImportIDSet();
      Thread.setDefaultUncaughtExceptionHandler(
             new DefaultExceptionHandler());
      indexMgr.setStarted();
      Message message =
              NOTE_JEB_IMPORT_LDIF_INDEX_STARTED.get(indexMgr.getFileName(),
                         indexMgr.getBufferList().size());
                         indexMgr.bufferIndexCount);
      logError(message);
      Integer cIndexID = null;

      ByteBuffer key = null;
      ImportIDSet insertIDSet = null;
      ImportIDSet deleteIDSet = null;
      Integer indexID = null;

      try
      {
        indexMgr.openIndexFile();
        SortedSet<Buffer> bufferSet = initializeBuffers();
        while(!bufferSet.isEmpty())
        NavigableSet<IndexInputBuffer> bufferSet = initializeBuffers();
        while (!bufferSet.isEmpty())
        {
          Buffer b;
          b = bufferSet.first();
          bufferSet.remove(b);
          if(cKey == null)
          IndexInputBuffer b = bufferSet.pollFirst();
          if (key == null)
          {
            cKey = ByteBuffer.allocate(BYTE_BUFFER_CAPACITY);
            cIndexID =  b.getIndexID();
            cKey.clear();
            if(b.getKeyLen() > cKey.capacity())
            indexID = b.getIndexID();

            if (indexMgr.isDN2ID())
            {
              cKey = ByteBuffer.allocate(b.getKeyLen());
            }
            cKey.flip();
            b.getKey(cKey);
            cInsertIDSet.merge(b.getInsertIDSet());
            cDeleteIDSet.merge(b.getDeleteIDSet());
            cInsertIDSet.setKey(cKey);
            cDeleteIDSet.setKey(cKey);
          }
          else
          {
            if(b.compare(cKey, cIndexID) != 0)
            {
              addToDB(cInsertIDSet, cDeleteIDSet, cIndexID);
              indexMgr.incrementKeyCount();
              cIndexID =  b.getIndexID();
              cKey.clear();
              if(b.getKeyLen() > cKey.capacity())
              {
                 cKey = ByteBuffer.allocate(b.getKeyLen());
              }
              cKey.flip();
              b.getKey(cKey);
              cInsertIDSet.clear(true);
              cDeleteIDSet.clear(true);
              cInsertIDSet.merge(b.getInsertIDSet());
              cDeleteIDSet.merge(b.getDeleteIDSet());
              cInsertIDSet.setKey(cKey);
              cDeleteIDSet.setKey(cKey);
              insertIDSet = new ImportIDSet(1, 1, false);
              deleteIDSet = new ImportIDSet(1, 1, false);
            }
            else
            {
              cInsertIDSet.merge(b.getInsertIDSet());
              cDeleteIDSet.merge(b.getDeleteIDSet());
              Index index = (Index) idContainerMap.get(indexID);
              int limit = index.getIndexEntryLimit();
              boolean doCount = index.getMaintainCount();
              insertIDSet = new ImportIDSet(1, limit, doCount);
              deleteIDSet = new ImportIDSet(1, limit, doCount);
            }

            key = ByteBuffer.allocate(b.getKeyLen());
            key.flip();
            b.getKey(key);

            b.mergeIDSet(insertIDSet);
            b.mergeIDSet(deleteIDSet);
            insertIDSet.setKey(key);
            deleteIDSet.setKey(key);
          }
          else if (b.compare(key, indexID) != 0)
          {
            addToDB(insertIDSet, deleteIDSet, indexID);
            indexMgr.incrementKeyCount();

            indexID = b.getIndexID();

            if (indexMgr.isDN2ID())
            {
              insertIDSet = new ImportIDSet(1, 1, false);
              deleteIDSet = new ImportIDSet(1, 1, false);
            }
            else
            {
              Index index = (Index) idContainerMap.get(indexID);
              int limit = index.getIndexEntryLimit();
              boolean doCount = index.getMaintainCount();
              insertIDSet = new ImportIDSet(1, limit, doCount);
              deleteIDSet = new ImportIDSet(1, limit, doCount);
            }

            key.clear();
            if (b.getKeyLen() > key.capacity())
            {
              key = ByteBuffer.allocate(b.getKeyLen());
            }
            key.flip();
            b.getKey(key);

            b.mergeIDSet(insertIDSet);
            b.mergeIDSet(deleteIDSet);
            insertIDSet.setKey(key);
            deleteIDSet.setKey(key);
          }
          else
          {
            b.mergeIDSet(insertIDSet);
            b.mergeIDSet(deleteIDSet);
          }

          if(b.hasMoreData())
          {
            b.getNextRecord();
            bufferSet.add(b);
          }
        }
        if(cKey != null)

        if(key != null)
        {
          addToDB(cInsertIDSet, cDeleteIDSet, cIndexID);
          addToDB(insertIDSet, deleteIDSet, indexID);
        }
      }
      catch (Exception e)
@@ -1986,7 +2071,7 @@

      private ByteBuffer parentDN, lastDN;
      private EntryID parentID, lastID, entryID;
      private final DatabaseEntry DNKey, DNValue;
      private final DatabaseEntry dnKey, dnValue;
      private final TreeMap<ByteBuffer, EntryID> parentIDMap;
      private final EntryContainer entryContainer;
      private final Map<byte[], ImportIDSet> id2childTree;
@@ -2009,8 +2094,8 @@
        subTreeLimit = entryContainer.getID2Subtree().getIndexEntryLimit();
        subTreeDoCount = entryContainer.getID2Subtree().getMaintainCount();
        id2subtreeTree =  new TreeMap<byte[], ImportIDSet>(subComparator);
        DNKey = new DatabaseEntry();
        DNValue = new DatabaseEntry();
        dnKey = new DatabaseEntry();
        dnValue = new DatabaseEntry();
        lastDN = ByteBuffer.allocate(BYTE_BUFFER_CAPACITY);
      }

@@ -2052,10 +2137,10 @@
      private boolean checkParent(ImportIDSet record) throws DirectoryException,
              DatabaseException
      {
        DNKey.setData(record.getKey().array(), 0 , record.getKey().limit());
        dnKey.setData(record.getKey().array(), 0 , record.getKey().limit());
        byte[] v = record.toDatabase();
        long v1 = JebFormat.entryIDFromDatabase(v);
        DNValue.setData(v);
        dnValue.setData(v);

        entryID = new EntryID(v1);
        parentDN = getParent(record.getKey());
@@ -2068,7 +2153,8 @@
          //If null is returned than this is a suffix DN.
          if(parentDN != null)
          {
            DatabaseEntry key = new DatabaseEntry(parentDN.array());
            DatabaseEntry key =
              new DatabaseEntry(parentDN.array(), 0 , parentDN.limit());
            DatabaseEntry value = new DatabaseEntry();
            OperationStatus status;
            status =
@@ -2166,7 +2252,7 @@
        if (importConfiguration != null &&
            importConfiguration.appendToExistingData())
        {
            DatabaseEntry key = new DatabaseEntry(dn.array());
            DatabaseEntry key = new DatabaseEntry(dn.array(), 0, dn.limit());
            DatabaseEntry value = new DatabaseEntry();
            OperationStatus status;
            status =
@@ -2235,7 +2321,7 @@

      public void writeToDB() throws DatabaseException, DirectoryException
      {
        entryContainer.getDN2ID().put(null, DNKey, DNValue);
        entryContainer.getDN2ID().put(null, dnKey, dnValue);
        indexMgr.addTotDNCount(1);
        if(parentDN != null)
        {
@@ -2253,8 +2339,8 @@
        {
          byte[] key = e.getKey();
          ImportIDSet idSet = e.getValue();
          DNKey.setData(key);
          index.insert(DNKey, idSet, DNValue);
          dnKey.setData(key);
          index.insert(dnKey, idSet, dnValue);
        }
        index.closeCursor();
        if(clearMap)
@@ -2281,7 +2367,7 @@
  {
    private final int DRAIN_TO = 3;
    private final IndexManager indexMgr;
    private final BlockingQueue<IndexBuffer> queue;
    private final BlockingQueue<IndexOutputBuffer> queue;
    private final ByteArrayOutputStream insetByteStream =
            new ByteArrayOutputStream(2 * bufferSize);
    private final ByteArrayOutputStream deleteByteStream =
@@ -2289,13 +2375,13 @@
    private final byte[] tmpArray = new byte[8];
    private int insertKeyCount = 0, deleteKeyCount = 0;
    private final DataOutputStream dataStream;
    private long bufferCount = 0;
    private int bufferCount = 0;
    private final File file;
    private final SortedSet<IndexBuffer> indexSortedSet;
    private final SortedSet<IndexOutputBuffer> indexSortedSet;
    private boolean poisonSeen = false;


    public ScratchFileWriterTask(BlockingQueue<IndexBuffer> queue,
    public ScratchFileWriterTask(BlockingQueue<IndexOutputBuffer> queue,
                             IndexManager indexMgr) throws FileNotFoundException
    {
      this.queue = queue;
@@ -2305,7 +2391,7 @@
              new BufferedOutputStream(new FileOutputStream(file),
                                       READER_WRITER_BUFFER_SIZE);
      dataStream = new DataOutputStream(bufferedStream);
      indexSortedSet = new TreeSet<IndexBuffer>();
      indexSortedSet = new TreeSet<IndexOutputBuffer>();
    }


@@ -2315,11 +2401,11 @@
    public Void call() throws IOException
    {
      long offset = 0;
      List<IndexBuffer> l = new LinkedList<IndexBuffer>();
      List<IndexOutputBuffer> l = new LinkedList<IndexOutputBuffer>();
      try {
        while(true)
        {
          IndexBuffer indexBuffer = queue.poll();
          IndexOutputBuffer indexBuffer = queue.poll();
          if(indexBuffer != null)
          {
            long beginOffset = offset;
@@ -2329,7 +2415,7 @@
              queue.drainTo(l, DRAIN_TO);
              l.add(indexBuffer);
              bufferLen = writeIndexBuffers(l);
              for(IndexBuffer id : l)
              for(IndexOutputBuffer id : l)
              {
                if(!id.isDiscard())
                {
@@ -2353,7 +2439,7 @@
              }
            }
            offset += bufferLen;
            indexMgr.addBuffer(new Buffer(beginOffset, offset, bufferCount));
            indexMgr.addBuffer(beginOffset, offset, bufferCount);
            bufferCount++;
            Importer.this.bufferCount.incrementAndGet();
            if(poisonSeen)
@@ -2381,7 +2467,8 @@
    }


    private long writeIndexBuffer(IndexBuffer indexBuffer) throws IOException
    private long writeIndexBuffer(IndexOutputBuffer indexBuffer)
      throws IOException
    {
      int numberKeys = indexBuffer.getNumberKeys();
      indexBuffer.setPosition(-1);
@@ -2433,14 +2520,14 @@
    }


    private long writeIndexBuffers(List<IndexBuffer> buffers)
    private long writeIndexBuffers(List<IndexOutputBuffer> buffers)
            throws IOException
    {
      long id = 0;
      long bufferLen = 0;
      insetByteStream.reset(); insertKeyCount = 0;
      deleteByteStream.reset(); deleteKeyCount = 0;
      for(IndexBuffer b : buffers)
      for(IndexOutputBuffer b : buffers)
      {
        if(b.isPoison())
        {
@@ -2457,7 +2544,7 @@
      int saveIndexID = 0;
      while(!indexSortedSet.isEmpty())
      {
        IndexBuffer b = indexSortedSet.first();
        IndexOutputBuffer b = indexSortedSet.first();
        indexSortedSet.remove(b);
        if(saveKey == null)
        {
@@ -2563,7 +2650,7 @@
    }


    private int writeRecord(IndexBuffer b) throws IOException
    private int writeRecord(IndexOutputBuffer b) throws IOException
    {
      int keySize = b.getKeySize();
      int packedSize = writeHeader(b.getIndexID(), keySize);
@@ -2595,9 +2682,9 @@
  private final class SortTask implements Callable<Void>
  {

    private final IndexBuffer indexBuffer;
    private final IndexOutputBuffer indexBuffer;

    public SortTask(IndexBuffer indexBuffer)
    public SortTask(IndexOutputBuffer indexBuffer)
    {
      this.indexBuffer = indexBuffer;
    }
@@ -2607,23 +2694,24 @@
     */
    public Void call() throws Exception
    {
      if (importConfiguration != null &&
          importConfiguration.isCancelled() || isPhaseOneCanceled)
      if (importConfiguration != null && importConfiguration.isCancelled()
          || isPhaseOneCanceled)
      {
        isPhaseOneCanceled =true;
        isPhaseOneCanceled = true;
        return null;
      }
      indexBuffer.sort();
      if(indexKeyQueMap.containsKey(indexBuffer.getIndexKey())) {
        BlockingQueue<IndexBuffer> q =
                indexKeyQueMap.get(indexBuffer.getIndexKey());
      if (indexKeyQueMap.containsKey(indexBuffer.getIndexKey()))
      {
        BlockingQueue<IndexOutputBuffer> q = indexKeyQueMap.get(indexBuffer
            .getIndexKey());
        q.add(indexBuffer);
      }
      else
      {
        createIndexWriterTask(indexBuffer.getIndexKey());
        BlockingQueue<IndexBuffer> q =
                                 indexKeyQueMap.get(indexBuffer.getIndexKey());
        BlockingQueue<IndexOutputBuffer> q = indexKeyQueMap.get(indexBuffer
            .getIndexKey());
        q.add(indexBuffer);
      }
      return null;
@@ -2653,8 +2741,8 @@
        {
          indexMgrList.add(indexMgr);
        }
        BlockingQueue<IndexBuffer> newQue =
                new ArrayBlockingQueue<IndexBuffer>(phaseOneBufferCount);
        BlockingQueue<IndexOutputBuffer> newQue =
                new ArrayBlockingQueue<IndexOutputBuffer>(phaseOneBufferCount);
        ScratchFileWriterTask indexWriter =
                new ScratchFileWriterTask(newQue, indexMgr);
        scratchFileWriterList.add(indexWriter);
@@ -2666,327 +2754,6 @@
  }

  /**
   * The buffer class is used to process a buffer from the temporary index files
   * during phase 2 processing.
   */
  private final class Buffer implements Comparable<Buffer>
  {
    private IndexManager indexMgr;
    private final long begin, end, id;
    private long offset;
    private ByteBuffer cache;
    private int limit;
    private ImportIDSet insertIDSet = null, deleteIDSet = null;
    private Integer indexID = null;
    private boolean doCount;
    private ByteBuffer keyBuf = ByteBuffer.allocate(BYTE_BUFFER_CAPACITY);


    public Buffer(long begin, long end, long id)
    {
      this.begin = begin;
      this.end = end;
      this.offset = 0;
      this.id = id;
    }


    private void initializeCache(IndexManager indexMgr, ByteBuffer b,
                      long cacheSize) throws IOException
    {
      this.indexMgr = indexMgr;
      if(b == null)
      {
        cache = ByteBuffer.allocate((int)cacheSize);
      }
      else
      {
        cache = b;
      }
      loadCache();
      cache.flip();
      keyBuf.flip();
    }


    private void loadCache() throws IOException
    {
      FileChannel fileChannel = indexMgr.getChannel();
      fileChannel.position(begin + offset);
      long leftToRead =  end - (begin + offset);
      long bytesToRead;
      if(leftToRead < cache.remaining())
      {
        cache.limit((int) (cache.position() + leftToRead));
        bytesToRead = (int)leftToRead;
      }
      else
      {
        bytesToRead = Math.min((end - offset),cache.remaining());
      }
      int bytesRead = 0;
      while(bytesRead < bytesToRead)
      {
        bytesRead += fileChannel.read(cache);
      }
      offset += bytesRead;
      indexMgr.addBytesRead(bytesRead);
    }

      public boolean hasMoreData() throws IOException
      {
        boolean ret = ((begin + offset) >= end);
        return !(cache.remaining() == 0 && ret);
      }

    public int getKeyLen()
    {
      return keyBuf.limit();
    }

    public void getKey(ByteBuffer b)
    {
      keyBuf.get(b.array(), 0, keyBuf.limit());
      b.limit(keyBuf.limit());
    }

    ByteBuffer getKeyBuf()
    {
      return keyBuf;
    }

    public ImportIDSet getInsertIDSet()
    {
      return insertIDSet;
    }

    public ImportIDSet getDeleteIDSet()
    {
      return deleteIDSet;
    }

    public long getBufferID()
    {
      return id;
    }

    public Integer getIndexID()
    {
      if(indexID == null)
      {
        try {
          getNextRecord();
        } catch(IOException ex) {
           Message message =
               ERR_JEB_IMPORT_BUFFER_IO_ERROR.get(indexMgr.getFileName());
           logError(message);
           ex.printStackTrace();
           System.exit(1);
        }
      }
      return indexID;
    }

    public void getNextRecord()  throws IOException
    {
      getNextIndexID();
      getContainerParameters();
      getNextKey();
      getNextIDSet(true);  //get insert ids
      getNextIDSet(false); //get delete ids
    }

    private void getContainerParameters()
    {
      limit = 1;
      doCount = false;
      if(!indexMgr.isDN2ID())
      {
        Index index = (Index) idContainerMap.get(indexID);
        limit = index.getIndexEntryLimit();
        doCount = index.getMaintainCount();
        if(insertIDSet == null)
        {
          insertIDSet = new ImportIDSet(128, limit, doCount);
          deleteIDSet = new ImportIDSet(128, limit, doCount);
        }
      }
      else
      {
        if(insertIDSet == null)
        {
            insertIDSet = new ImportIDSet(1, limit, doCount);
            deleteIDSet = new ImportIDSet(1, limit, doCount);
        }
      }
    }


    private int getInt()  throws IOException
    {
      ensureData(4);
      return cache.getInt();
    }

    private void getNextIndexID() throws IOException, BufferUnderflowException
     {
       indexID = getInt();
     }

    private void getNextKey() throws IOException, BufferUnderflowException
    {
      ensureData(20);
      byte[] ba = cache.array();
      int p = cache.position();
      int len = PackedInteger.getReadIntLength(ba, p);
      int keyLen = PackedInteger.readInt(ba, p);
      cache.position(p + len);
      if(keyLen > keyBuf.capacity())
      {
        keyBuf = ByteBuffer.allocate(keyLen);
      }
      ensureData(keyLen);
      keyBuf.clear();
      cache.get(keyBuf.array(), 0, keyLen);
      keyBuf.limit(keyLen);
    }

    private void getNextIDSet(boolean insert)
            throws IOException, BufferUnderflowException
    {
      ensureData(20);
      int p = cache.position();
      byte[] ba = cache.array();
      int len = PackedInteger.getReadIntLength(ba, p);
      int keyCount = PackedInteger.readInt(ba, p);
      p += len;
      cache.position(p);
      if(insert)
      {
        insertIDSet.clear(false);
      }
      else
      {
        deleteIDSet.clear(false);
      }
      for(int k = 0; k < keyCount; k++)
      {
        if(ensureData(9))
        {
          p = cache.position();
        }
        len = PackedInteger.getReadLongLength(ba, p);
        long l = PackedInteger.readLong(ba, p);
        p += len;
        cache.position(p);
        if(insert)
        {
          insertIDSet.addEntryID(l);
        }
        else
        {
          deleteIDSet.addEntryID(l);
        }
      }
    }


    private boolean ensureData(int len) throws IOException
    {
      boolean ret = false;
      if(cache.remaining() == 0)
      {
        cache.clear();
        loadCache();
        cache.flip();
        ret = true;
      }
      else if(cache.remaining() < len)
      {
        cache.compact();
        loadCache();
        cache.flip();
        ret = true;
      }
      return ret;
    }


    private int compare(ByteBuffer cKey, Integer cIndexID)
    {
      int returnCode, rc;
      if(keyBuf.limit() == 0)
      {
        getIndexID();
      }
      rc = indexComparator.compare(keyBuf.array(), 0, keyBuf.limit(),
                                     cKey.array(), cKey.limit());
      if(rc != 0) {
        returnCode = 1;
      }
      else
      {
        returnCode = (indexID.intValue() == cIndexID.intValue()) ? 0 : 1;
      }
      return returnCode;
    }



    public int compareTo(Buffer o) {
      //used in remove.
      if(this.equals(o))
      {
        return 0;
      }
      if(keyBuf.limit() == 0) {
        getIndexID();
      }
      if(o.getKeyBuf().limit() == 0)
      {
        o.getIndexID();
      }
      int returnCode;
      byte[] oKey = o.getKeyBuf().array();
      int oLen = o.getKeyBuf().limit();
      returnCode = indexComparator.compare(keyBuf.array(), 0, keyBuf.limit(),
                                             oKey, oLen);
      if(returnCode == 0)
      {
        if(indexID.intValue() == o.getIndexID().intValue())
        {
          if(insertIDSet.isDefined())
          {
            returnCode = -1;
          }
          else if(o.getInsertIDSet().isDefined())
          {
            returnCode = 1;
          }
          else if(insertIDSet.size() == o.getInsertIDSet().size())
          {
            returnCode = id > o.getBufferID() ? 1 : -1;
          }
          else
          {
            returnCode = insertIDSet.size() - o.getInsertIDSet().size();
          }
        }
        else if(indexID > o.getIndexID())
        {
          returnCode = 1;
        }
        else
        {
          returnCode = -1;
        }
      }
      return returnCode;
    }
  }


  /**
   * The index manager class has several functions:
   *
   *   1. It used to carry information about index processing created in phase
@@ -2996,11 +2763,12 @@
   *
   *   3. It manages opening and closing the scratch index files.
   */
  private final class IndexManager implements Comparable<IndexManager>
  final class IndexManager implements Comparable<IndexManager>
  {
    private static final int BUFFER_SIZE = 128;

    private final File file;
    private RandomAccessFile rFile = null;
    private final List<Buffer> bufferList = new LinkedList<Buffer>();
    private long fileLength, bytesRead = 0;
    private boolean done = false, started = false;
    private long totalDNS;
@@ -3009,8 +2777,12 @@
    private final boolean isDN;
    private final int limit;

    private long[] bufferIndexBegin = new long[BUFFER_SIZE];
    private long[] bufferIndexEnd   = new long[BUFFER_SIZE];
    private int[]  bufferIndexID    = new int[BUFFER_SIZE];
    private int    bufferIndexCount = 0;

    IndexManager(String fileName, boolean isDN, int limit)
    private IndexManager(String fileName, boolean isDN, int limit)
    {
      file = new File(tempDir, fileName);
      this.fileName = fileName;
@@ -3019,91 +2791,110 @@
    }


    void openIndexFile() throws FileNotFoundException
    private void openIndexFile() throws FileNotFoundException
    {
      rFile = new RandomAccessFile(file, "r");
    }


    public FileChannel getChannel()
    /**
     * Returns the file channel associated with this index manager.
     *
     * @return The file channel associated with this index manager.
     */
    FileChannel getChannel()
    {
      return rFile.getChannel();
    }


    public void addBuffer(Buffer o)

    private void addBuffer(long begin, long end, int id)
    {
      this.bufferList.add(o);
      int size = bufferIndexBegin.length;
      if (bufferIndexCount >= size)
      {
        size += BUFFER_SIZE;
        bufferIndexBegin = Arrays.copyOf(bufferIndexBegin, size);
        bufferIndexEnd = Arrays.copyOf(bufferIndexEnd, size);
        bufferIndexID = Arrays.copyOf(bufferIndexID, size);
      }
      bufferIndexBegin[bufferIndexCount] = begin;
      bufferIndexEnd[bufferIndexCount] = end;
      bufferIndexID[bufferIndexCount] = id;
      bufferIndexCount++;
    }


    public List<Buffer> getBufferList()
    {
      return bufferList;
    }


    public File getFile()
    private File getFile()
    {
      return file;
    }


    public boolean deleteIndexFile()
    private boolean deleteIndexFile()
    {
      return file.delete();
    }


    public void close() throws IOException
    private void close() throws IOException
    {
      rFile.close();
    }


    public void setFileLength()
    private void setFileLength()
    {
      this.fileLength = file.length();
    }


    public void addBytesRead(int bytesRead)

    /**
     * Updates the bytes read counter.
     *
     * @param bytesRead
     *          The number of bytes read.
     */
    void addBytesRead(int bytesRead)
    {
      this.bytesRead += bytesRead;
    }


    public void setDone()
    private void setDone()
    {
      this.done = true;
    }


    public void setStarted()
    private void setStarted()
    {
      started = true;
    }


    public void addTotDNCount(int delta)
    private void addTotDNCount(int delta)
    {
      this.totalDNS += delta;
    }


    public long getDNCount()
    private long getDNCount()
    {
      return totalDNS;
    }


    public boolean isDN2ID()
    private boolean isDN2ID()
    {
      return isDN;
    }


    public void printStats(long deltaTime)
    private void printStats(long deltaTime)
    {
      if(!done && started)
      {
@@ -3115,38 +2906,35 @@
    }


    public void incrementKeyCount()
    private void incrementKeyCount()
    {
      keyCount.incrementAndGet();
    }


    public String getFileName()
    /**
     * Returns the file name associated with this index manager.
     *
     * @return The file name associated with this index manager.
     */
    String getFileName()
    {
      return fileName;
    }


    public int getLimit()
    private int getLimit()
    {
      return limit;
    }


    /**
     * {@inheritDoc}
     */
    public int compareTo(IndexManager mgr)
    {
      if(bufferList.size() == mgr.getBufferList().size())
      {
         return 0;
      }
      else if (bufferList.size() < mgr.getBufferList().size())
      {
        return -1;
      }
      else
      {
        return 1;
      }
      return bufferIndexCount - mgr.bufferIndexCount;
    }
  }

@@ -3154,7 +2942,7 @@
  /**
   * The rebuild index manager handles all rebuild index related processing.
   */
  class RebuildIndexManager extends ImportTask {
  private class RebuildIndexManager extends ImportTask {

   //Rebuild index configuration.
   private final RebuildConfig rebuildConfig;
@@ -3320,24 +3108,32 @@
    }



    /**
     * Perform rebuild index processing.
     *
     * @throws DatabaseException If an database error occurred.
     * @throws InterruptedException If an interrupted error occurred.
     * @throws ExecutionException If an Excecution error occurred.
     * @throws JebException If an JEB error occurred.
     * @throws InitializationException
     *           If an initialization error occurred.
     * @throws DatabaseException
     *           If an database error occurred.
     * @throws InterruptedException
     *           If an interrupted error occurred.
     * @throws ExecutionException
     *           If an Excecution error occurred.
     * @throws JebException
     *           If an JEB error occurred.
     */
    public void rebuldIndexes() throws DatabaseException, InterruptedException,
            ExecutionException, JebException
    public void rebuldIndexes() throws InitializationException,
        DatabaseException, InterruptedException, ExecutionException,
        JebException
    {
      phaseOne();
      if(isPhaseOneCanceled)
      if (isPhaseOneCanceled)
      {
        throw new InterruptedException("Rebuild Index canceled.");
      }
      phaseTwo();
      if(rebuildAll)
      if (rebuildAll)
      {
        setAllIndexesTrusted();
      }
@@ -3455,11 +3251,12 @@
    }


    private void phaseTwo() throws InterruptedException, JebException,
            ExecutionException

    private void phaseTwo() throws InitializationException,
        InterruptedException, JebException, ExecutionException
    {
      SecondPhaseProgressTask progressTask =
              new SecondPhaseProgressTask(entriesProcessed.get());
      SecondPhaseProgressTask progressTask = new SecondPhaseProgressTask(
          entriesProcessed.get());
      Timer timer2 = new Timer();
      timer2.scheduleAtFixedRate(progressTask, TIMER_INTERVAL, TIMER_INTERVAL);
      processIndexFiles();
@@ -4061,7 +3858,7 @@
   * This class reports progress of rebuild index processing at fixed
   * intervals.
   */
  class RebuildFirstPhaseProgressTask extends TimerTask
  private class RebuildFirstPhaseProgressTask extends TimerTask
  {
    /**
     * The number of records that had been processed at the time of the
@@ -4299,7 +4096,7 @@
   * This class reports progress of the second phase of import processing at
   * fixed intervals.
   */
  class SecondPhaseProgressTask extends TimerTask
  private class SecondPhaseProgressTask extends TimerTask
  {
    /**
     * The number of entries that had been read at the time of the

 opends/src/server/org/opends/server/backends/jeb/importLDIF/IndexInputBuffer.java

New file
@@ -0,0 +1,414 @@
/*
 * CDDL HEADER START
 *
 * The contents of this file are subject to the terms of the
 * Common Development and Distribution License, Version 1.0 only
 * (the "License").  You may not use this file except in compliance
 * with the License.
 *
 * You can obtain a copy of the license at
 * trunk/opends/resource/legal-notices/OpenDS.LICENSE
 * or https://OpenDS.dev.java.net/OpenDS.LICENSE.
 * See the License for the specific language governing permissions
 * and limitations under the License.
 *
 * When distributing Covered Code, include this CDDL HEADER in each
 * file and include the License file at
 * trunk/opends/resource/legal-notices/OpenDS.LICENSE.  If applicable,
 * add the following below this CDDL HEADER, with the fields enclosed
 * by brackets "[]" replaced with your own identifying information:
 *      Portions Copyright [yyyy] [name of copyright owner]
 *
 * CDDL HEADER END
 *
 *
 *      Copyright 2010 Sun Microsystems, Inc.
 */

package org.opends.server.backends.jeb.importLDIF;



import static org.opends.messages.JebMessages.ERR_JEB_IMPORT_BUFFER_IO_ERROR;
import static org.opends.server.loggers.ErrorLogger.logError;

import java.io.IOException;
import java.nio.ByteBuffer;
import java.nio.channels.FileChannel;

import org.opends.messages.Message;
import org.opends.server.backends.jeb.importLDIF.Importer.IndexManager;

import com.sleepycat.util.PackedInteger;



/**
 * The buffer class is used to process a buffer from the temporary index files
 * during phase 2 processing.
 */
public final class IndexInputBuffer implements Comparable<IndexInputBuffer>
{
  private final IndexManager indexMgr;
  private final long begin;
  private final long end;
  private final int id;

  private long offset;
  private ByteBuffer cache;
  private Integer indexID = null;
  private ByteBuffer keyBuf = ByteBuffer.allocate(128);



  /**
   * Possible states while reading a record.
   */
  private enum RecordState
  {
    START, NEED_INSERT_ID_SET, NEED_DELETE_ID_SET
  };



  private RecordState recordState = RecordState.START;



  /**
   * Creates a new index input buffer.
   *
   * @param indexMgr
   *          The index manager.
   * @param begin
   *          The position of the start of the buffer in the scratch file.
   * @param end
   *          The position of the end of the buffer in the scratch file.
   * @param id
   *          The index ID.
   */
  public IndexInputBuffer(IndexManager indexMgr, long begin, long end, int id)
  {
    this.indexMgr = indexMgr;
    this.begin = begin;
    this.end = end;
    this.offset = 0;
    this.id = id;
  }



  /**
   * Initializes this index input buffer using the provided cache size.
   *
   * @param cacheSize
   *          The cache size.
   * @throws IOException
   *           If an IO error occurred.
   */
  void initializeCache(long cacheSize) throws IOException
  {
    cache = ByteBuffer.allocate((int) Math.max(cacheSize - 256, 256));
    loadCache();
    cache.flip();
    keyBuf.flip();
  }



  private void loadCache() throws IOException
  {
    FileChannel fileChannel = indexMgr.getChannel();
    fileChannel.position(begin + offset);
    long leftToRead = end - (begin + offset);
    long bytesToRead;
    if (leftToRead < cache.remaining())
    {
      cache.limit((int) (cache.position() + leftToRead));
      bytesToRead = (int) leftToRead;
    }
    else
    {
      bytesToRead = Math.min((end - offset), cache.remaining());
    }
    int bytesRead = 0;
    while (bytesRead < bytesToRead)
    {
      bytesRead += fileChannel.read(cache);
    }
    offset += bytesRead;
    indexMgr.addBytesRead(bytesRead);
  }



  /**
   * Returns {@code true} if this buffer has more data.
   *
   * @return {@code true} if this buffer has more data.
   * @throws IOException
   *           If an IO error occurred.
   */
  public boolean hasMoreData() throws IOException
  {
    boolean ret = ((begin + offset) >= end);
    return !(cache.remaining() == 0 && ret);
  }



  /**
   * Returns the length of the next key.
   *
   * @return The length of the next key.
   */
  public int getKeyLen()
  {
    return keyBuf.limit();
  }



  /**
   * Returns the next key.
   *
   * @param b
   *          A buffer into which the key should be added.
   */
  public void getKey(ByteBuffer b)
  {
    keyBuf.get(b.array(), 0, keyBuf.limit());
    b.limit(keyBuf.limit());
  }



  /**
   * Returns the index ID of the next record.
   *
   * @return The index ID of the next record.
   */
  public Integer getIndexID()
  {
    if (indexID == null)
    {
      try
      {
        getNextRecord();
      }
      catch (IOException ex)
      {
        Message message = ERR_JEB_IMPORT_BUFFER_IO_ERROR.get(indexMgr
            .getFileName());
        logError(message);
        ex.printStackTrace();
        System.exit(1);
      }
    }
    return indexID;
  }



  /**
   * Reads the next record from the buffer, skipping any remaining data in the
   * current record.
   *
   * @throws IOException
   *           If an IO error occurred.
   */
  public void getNextRecord() throws IOException
  {
    switch (recordState)
    {
    case START:
      // Nothing to skip.
      break;
    case NEED_INSERT_ID_SET:
      // The previous record's ID sets were not read, so skip them both.
      mergeIDSet(null);
      mergeIDSet(null);
      break;
    case NEED_DELETE_ID_SET:
      // The previous record's delete ID set was not read, so skip it.
      mergeIDSet(null);
      break;
    }

    indexID = getInt();

    ensureData(20);
    byte[] ba = cache.array();
    int p = cache.position();
    int len = PackedInteger.getReadIntLength(ba, p);
    int keyLen = PackedInteger.readInt(ba, p);
    cache.position(p + len);
    if (keyLen > keyBuf.capacity())
    {
      keyBuf = ByteBuffer.allocate(keyLen);
    }
    ensureData(keyLen);
    keyBuf.clear();
    cache.get(keyBuf.array(), 0, keyLen);
    keyBuf.limit(keyLen);

    recordState = RecordState.NEED_INSERT_ID_SET;
  }



  private int getInt() throws IOException
  {
    ensureData(4);
    return cache.getInt();
  }



  /**
   * Reads the next ID set from the record and merges it with the provided ID
   * set.
   *
   * @param idSet
   *          The ID set to be merged.
   * @throws IOException
   *           If an IO error occurred.
   */
  public void mergeIDSet(ImportIDSet idSet) throws IOException
  {
    if (recordState == RecordState.START)
    {
      throw new IllegalStateException();
    }

    ensureData(20);
    int p = cache.position();
    byte[] ba = cache.array();
    int len = PackedInteger.getReadIntLength(ba, p);
    int keyCount = PackedInteger.readInt(ba, p);
    p += len;
    cache.position(p);
    for (int k = 0; k < keyCount; k++)
    {
      if (ensureData(9))
      {
        p = cache.position();
      }
      len = PackedInteger.getReadLongLength(ba, p);
      long l = PackedInteger.readLong(ba, p);
      p += len;
      cache.position(p);

      // idSet will be null if skipping.
      if (idSet != null)
      {
        idSet.addEntryID(l);
      }
    }

    switch (recordState)
    {
    case START:
      throw new IllegalStateException();
    case NEED_INSERT_ID_SET:
      recordState = RecordState.NEED_DELETE_ID_SET;
      break;
    case NEED_DELETE_ID_SET:
      recordState = RecordState.START;
      break;
    }
  }



  private boolean ensureData(int len) throws IOException
  {
    boolean ret = false;
    if (cache.remaining() == 0)
    {
      cache.clear();
      loadCache();
      cache.flip();
      ret = true;
    }
    else if (cache.remaining() < len)
    {
      cache.compact();
      loadCache();
      cache.flip();
      ret = true;
    }
    return ret;
  }



  /**
   * Compares this buffer with the provided key and index ID.
   *
   * @param cKey
   *          The key.
   * @param cIndexID
   *          The index ID.
   * @return A negative number if this buffer is less than the provided key and
   *         index ID, a positive number if this buffer is greater, or zero if
   *         it is the same.
   */
  int compare(ByteBuffer cKey, Integer cIndexID)
  {
    int returnCode, rc;
    if (keyBuf.limit() == 0)
    {
      getIndexID();
    }
    rc = Importer.indexComparator.compare(keyBuf.array(), 0, keyBuf.limit(),
        cKey.array(), cKey.limit());
    if (rc != 0)
    {
      returnCode = 1;
    }
    else
    {
      returnCode = (indexID.intValue() == cIndexID.intValue()) ? 0 : 1;
    }
    return returnCode;
  }



  /**
   * {@inheritDoc}
   */
  public int compareTo(IndexInputBuffer o)
  {
    // used in remove.
    if (this == o)
    {
      return 0;
    }

    if (keyBuf.limit() == 0)
    {
      getIndexID();
    }

    if (o.keyBuf.limit() == 0)
    {
      o.getIndexID();
    }

    byte[] oKey = o.keyBuf.array();
    int oLen = o.keyBuf.limit();
    int returnCode = Importer.indexComparator.compare(keyBuf.array(), 0,
        keyBuf.limit(), oKey, oLen);
    if (returnCode == 0)
    {
      returnCode = indexID.intValue() - o.getIndexID().intValue();
      if (returnCode == 0)
      {
        returnCode = id - o.id;
      }
    }
    return returnCode;
  }
}

 opends/src/server/org/opends/server/backends/jeb/importLDIF/IndexOutputBuffer.java

File was renamed from opends/src/server/org/opends/server/backends/jeb/importLDIF/IndexBuffer.java
@@ -51,7 +51,7 @@
 * This class is not thread safe.
 *
 */
public class IndexBuffer implements Comparable<IndexBuffer> {
public final class IndexOutputBuffer implements Comparable<IndexOutputBuffer> {

  /**
  * Enumeration used when sorting a buffer.
@@ -110,23 +110,16 @@
  private boolean discard = false;


  private IndexBuffer(int size) {
    this.size = size;
    this.buffer = new byte[size];
    this.bytesLeft = size;
    this.recordOffset = size - 1;
  }


  /**
   * Create an instance of a IndexBuffer using the specified size.
   *
   * @param size The size of the underlying byte array.
   * @return A newly created instance of an IndexBuffer.
   */
  public static
  IndexBuffer createIndexBuffer(int size) {
    return new IndexBuffer(size);
  public IndexOutputBuffer(int size) {
    this.size = size;
    this.buffer = new byte[size];
    this.bytesLeft = size;
    this.recordOffset = size - 1;
  }


@@ -160,7 +153,7 @@
   *
   * @return The value of a buffer's ID.
   */
  public long getBufferID()
  private long getBufferID()
  {
    return this.id;
  }
@@ -261,14 +254,14 @@
   * Add the specified key byte array and EntryID to the buffer.
   *
   * @param keyBytes The key byte array.
   * @param IDEntry The EntryID.
   * @param entryID The EntryID.
   * @param indexID The index ID the record belongs.
   * @param insert <CODE>True</CODE> if key is an insert, false otherwise.
   */

  public void add(byte[] keyBytes, EntryID IDEntry, int indexID,
  public void add(byte[] keyBytes, EntryID entryID, int indexID,
                  boolean insert) {
    recordOffset = addRecord(keyBytes, IDEntry.longValue(), indexID, insert);
    recordOffset = addRecord(keyBytes, entryID.longValue(), indexID, insert);
    System.arraycopy(getIntBytes(recordOffset), 0, buffer, keyOffset, 4);
    keyOffset += 4;
    bytesLeft = recordOffset - keyOffset;
@@ -488,7 +481,7 @@
   * @return  0 if the buffers are equal, -1 if the current buffer is less
   *          than the specified buffer, or 1 if it is greater.
   */
  public int compareTo(IndexBuffer b)
  public int compareTo(IndexOutputBuffer b)
  {
    ByteBuffer keyBuf = b.getKeyBuf(b.position);
    int offset = getIntegerValue(position * 4);
@@ -799,7 +792,7 @@
   * they are non-DN indexes.
   */
  public static
  class IndexComparator implements IndexBuffer.ComparatorBuffer<byte[]>
  class IndexComparator implements IndexOutputBuffer.ComparatorBuffer<byte[]>
  {

    /**

 opends/src/server/org/opends/server/backends/jeb/importLDIF/Suffix.java

@@ -22,7 +22,7 @@
 * CDDL HEADER END
 *
 *
 *      Copyright 2009 Sun Microsystems, Inc.
 *      Copyright 2009-2010 Sun Microsystems, Inc.
 */

package org.opends.server.backends.jeb.importLDIF;
@@ -54,7 +54,7 @@
  private final EntryContainer srcEntryContainer;
  private EntryContainer entryContainer;
  private final Object synchObject = new Object();
  private static final int PARENT_ID_SET_SIZE = 16 * KB;
  private static final int PARENT_ID_SET_SIZE = 16 * 1024;
  private ConcurrentHashMap<DN, CountDownLatch> pendingMap =
          new ConcurrentHashMap<DN, CountDownLatch>();
  private Set<DN> parentSet = new HashSet<DN>(PARENT_ID_SET_SIZE);

 opends/src/server/org/opends/server/tools/ImportLDIF.java

@@ -46,7 +46,6 @@
import org.opends.messages.Message;
import org.opends.server.admin.std.server.BackendCfg;
import org.opends.server.api.Backend;
import org.opends.server.api.DebugLogPublisher;
import org.opends.server.api.ErrorLogPublisher;
import org.opends.server.api.plugin.PluginType;
import org.opends.server.config.ConfigException;
@@ -58,8 +57,6 @@
import org.opends.server.loggers.ErrorLogger;
import org.opends.server.loggers.TextErrorLogPublisher;
import org.opends.server.loggers.TextWriter;
import org.opends.server.loggers.debug.DebugLogger;
import org.opends.server.loggers.debug.TextDebugLogPublisher;
import org.opends.server.protocols.ldap.LDAPAttribute;
import org.opends.server.tasks.ImportTask;
import org.opends.server.tools.makeldif.TemplateFile;
@@ -729,7 +726,7 @@
  /**
   * {@inheritDoc}
   */
  public Class getTaskClass() {
  public Class<?> getTaskClass() {
    return ImportTask.class;
  }

@@ -866,14 +863,10 @@
      {
        try
        {
          ErrorLogPublisher errorLogPublisher =
          ErrorLogPublisher<?> errorLogPublisher =
              TextErrorLogPublisher.getStartupTextErrorPublisher(
                  new TextWriter.STREAM(out));
          DebugLogPublisher debugLogPublisher =
              TextDebugLogPublisher.getStartupTextDebugPublisher(
                  new TextWriter.STREAM(out));
          ErrorLogger.addErrorLogPublisher(errorLogPublisher);
          DebugLogger.addDebugLogPublisher(debugLogPublisher);
        }
        catch(Exception e)
        {

 opends/src/server/org/opends/server/util/Platform.java

@@ -35,9 +35,13 @@
import java.security.cert.Certificate;
import java.security.cert.CertificateFactory;
import java.security.cert.X509Certificate;
import java.util.List;
import java.io.FileInputStream;
import java.io.FileOutputStream;
import java.io.InputStream;
import java.lang.management.ManagementFactory;
import java.lang.management.MemoryPoolMXBean;
import java.lang.management.MemoryUsage;
import java.lang.reflect.Constructor;
import java.lang.reflect.Method;
import org.opends.messages.Message;
@@ -391,6 +395,72 @@
     *          The buffer to normalize.
     */
    public abstract void normalize(StringBuilder buffer);



    /**
     * Calculates the usable memory which could potentially be used by the
     * application for caching objects.
     *
     * @return The usable memory which could potentially be used by the
     *         application for caching objects.
     */
    public long getUsableMemoryForCaching()
    {
      long youngGenSize = 0;
      long oldGenSize = 0;

      List<MemoryPoolMXBean> mpools = ManagementFactory.getMemoryPoolMXBeans();
      for (MemoryPoolMXBean mpool : mpools)
      {
        MemoryUsage usage = mpool.getUsage();
        if (usage != null)
        {
          String name = mpool.getName();
          if (name.equalsIgnoreCase("PS Eden Space"))
          {
            // Parallel.
            youngGenSize = usage.getMax();
          }
          else if (name.equalsIgnoreCase("PS Old Gen"))
          {
            // Parallel.
            oldGenSize = usage.getMax();
          }
          else if (name.equalsIgnoreCase("Par Eden Space"))
          {
            // CMS.
            youngGenSize = usage.getMax();
          }
          else if (name.equalsIgnoreCase("CMS Old Gen"))
          {
            // CMS.
            oldGenSize = usage.getMax();
          }
        }
      }

      if (youngGenSize > 0 && oldGenSize > youngGenSize)
      {
        // We can calculate available memory based on GC info.
        return oldGenSize - youngGenSize;
      }
      else if (oldGenSize > 0)
      {
        // Small old gen. It is going to be difficult to avoid full GCs if the
        // young gen is bigger.
        return oldGenSize * 40 / 100;
      }
      else
      {
        // Unknown GC (G1, JRocket, etc).
        Runtime runTime = Runtime.getRuntime();
        runTime.gc();
        runTime.gc();
        return (runTime.freeMemory() + (runTime.maxMemory() - runTime
            .totalMemory())) * 40 / 100;
      }
    }
  }


@@ -627,4 +697,32 @@
    String javaVendor = System.getProperty("java.vendor");
    return javaVendor.startsWith(vendor);
  }



  /**
   * Calculates the usable memory which could potentially be used by the
   * application for caching objects. This method <b>does not</b> look at the
   * amount of free memory, but instead tries to query the JVM's GC settings in
   * order to determine the amount of usable memory in the old generation (or
   * equivalent). More specifically, applications may also need to take into
   * account the amount of memory already in use, for example by performing the
   * following:
   *
   * <pre>
   * Runtime runTime = Runtime.getRuntime();
   * runTime.gc();
   * runTime.gc();
   * long freeCommittedMemory = runTime.freeMemory();
   * long uncommittedMemory = runTime.maxMemory() - runTime.totalMemory();
   * long freeMemory = freeCommittedMemory + uncommittedMemory;
   * </pre>
   *
   * @return The usable memory which could potentially be used by the
   *         application for caching objects.
   */
  public static long getUsableMemoryForCaching()
  {
    return IMPL.getUsableMemoryForCaching();
  }
}

 opends/tests/unit-tests-testng/src/server/org/opends/server/backends/jeb/TestImportJob.java

@@ -565,8 +565,7 @@
  }


  // Re-enable when 6962694 fixed.
  @Test(enabled=false, dependsOnMethods = "testImportReplaceExisting")
  @Test(dependsOnMethods = "testImportReplaceExisting")
  public void testImportAppend() throws Exception
  {
    TestCaseUtils.clearJEBackend(false, beID, null);