Importer.java revision 8ac57ee1cd50fcc3d02b36bea4ab1335924f1d7a
/*
* 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 legal-notices/CDDLv1_0.txt
* 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 legal-notices/CDDLv1_0.txt.
* 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 2008-2010 Sun Microsystems, Inc.
* Portions Copyright 2011-2015 ForgeRock AS
*/
/**
* This class provides the engine that performs both importing of LDIF files and
* the rebuilding of indexes.
*/
final class Importer
{
/**
* Shim that allows properly constructing an {@link Importer} without polluting
* {@link ImportStrategy} and {@link RootContainer} with this importer inner workings.
*/
static final class StrategyImpl implements ImportStrategy
{
private final PluggableBackendCfg backendCfg;
{
this.backendCfg = backendCfg;
}
{
try
{
}
catch (DirectoryException | InitializationException e)
{
logger.traceException(e);
throw e;
}
catch (ConfigException e)
{
logger.traceException(e);
}
catch (Exception e)
{
logger.traceException(e);
throw new DirectoryException(getServerErrorResultCode(),
}
}
}
private static final int TIMER_INTERVAL = 10000;
/** Defaults for DB cache. */
/**
* Defaults for LDIF reader buffers, min memory required to import and default
* size for byte buffers.
*/
/** Max size of phase one buffer. */
/** Min size of phase one buffer. */
/** Min size of phase two read-ahead cache. */
/** Small heap threshold used to give more memory to JVM to attempt OOM errors. */
/** The DN attribute type. */
private static final AttributeType DN_TYPE;
static
{
{
}
}
/** Root container. */
private final RootContainer rootContainer;
/** Import configuration. */
private final LDIFImportConfig importCfg;
private final ServerContext serverContext;
/** LDIF reader. */
private ImportLDIFReader reader;
/** Phase one buffer count. */
/** Phase one imported entries count. */
/** Migrated entry count. */
private int migratedCount;
/** Phase one buffer size in bytes. */
private int bufferSize;
/** Index count. */
private final int indexCount;
/** Thread count. */
private int threadCount;
/** Whether DN validation should be performed. If true, then it is performed during phase one. */
private final boolean validateDNs;
/** Temp scratch directory. */
/** DN cache used when DN validation is done in first phase. */
/** Size in bytes of DN cache. */
private long dnCacheSize;
/** Available memory at the start of the import. */
private long availableMemory;
/** Size in bytes of DB cache. */
private long dbCacheSize;
/** The executor service used for the buffer sort tasks. */
private ExecutorService bufferSortService;
/** The executor service used for the scratch file processing tasks. */
private ExecutorService scratchFileWriterService;
/** Queue of free index buffers -- used to re-cycle index buffers. */
/**
* Map of index keys to index buffers. Used to allocate sorted index buffers
* to a index writer thread.
*/
private final Map<IndexKey, BlockingQueue<IndexOutputBuffer>> indexKeyQueueMap = new ConcurrentHashMap<>();
/** Map of DB containers to index managers. Used to start phase 2. */
/** Map of DB containers to DN-based index managers. Used to start phase 2. */
/**
* Futures used to indicate when the index file writers are done flushing
* their work queues and have exited. End of phase one.
*/
/**
* List of index file writer tasks. Used to signal stopScratchFileWriters to
* the index file writer tasks when the LDIF file has been done.
*/
/** Map of DNs to Suffix objects. */
/**
* Map of indexIDs to indexes.
* <p>
* Mainly used to support multiple suffixes. Each index in each suffix gets a unique ID to
* identify which tree it needs to go to in phase two processing.
*/
/** Map of indexIDs to entry containers. */
private final ConcurrentHashMap<Integer, EntryContainer> indexIDToECMap = new ConcurrentHashMap<>();
/** Used to synchronize when a scratch file index writer is first setup. */
/** Rebuild index manager used when rebuilding indexes. */
private final RebuildIndexManager rebuildManager;
/** Set to true if the backend was cleared. */
private final boolean clearedBackend;
/** Used to shutdown import if an error occurs in phase one. */
private volatile boolean isCanceled;
/** Number of phase one buffers. */
private int phaseOneBufferCount;
/**
* Create a new import job with the specified rebuild index config.
*
* @param rebuildConfig
* The rebuild index configuration.
* @param cfg
* The backend configuration.
* @throws InitializationException
* If a problem occurs during initialization.
* @throws StorageRuntimeException
* If an error occurred when opening the DB.
* @throws ConfigException
* If a problem occurs during initialization.
*/
ServerContext serverContext) throws InitializationException, StorageRuntimeException, ConfigException
{
this.rootContainer = rootContainer;
this.serverContext = serverContext;
this.threadCount = 1;
this.clearedBackend = false;
this.validateDNs = false;
}
/**
* Create a new import job with the specified ldif import config.
*
* @param importCfg
* The LDIF import configuration.
* @param backendCfg
* The backend configuration.
* @throws InitializationException
* If a problem occurs during initialization.
* @throws ConfigException
* If a problem occurs reading the configuration.
* @throws StorageRuntimeException
* If an error occurred when opening the DB.
*/
ServerContext serverContext) throws InitializationException, ConfigException, StorageRuntimeException
{
this.rootContainer = rootContainer;
this.rebuildManager = null;
this.serverContext = serverContext;
{
}
else
{
}
// Determine the number of indexes.
// be careful: requires that a few data has been set
if (validateDNs)
{
}
else
{
}
}
private File prepareTempDir(PluggableBackendCfg backendCfg, String tmpDirectory) throws InitializationException
{
{
}
return tempDir;
}
/**
* Returns whether the backend must be cleared.
*
* @param importCfg
* the import configuration object
* @param backendCfg
* the backend configuration object
* @return true if the backend must be cleared, false otherwise
* @see Importer#prepareSuffix(WriteableTransaction, EntryContainer) for per-suffix cleanups.
*/
{
return !importCfg.appendToExistingData()
/*
* Why do we clear when there is only one baseDN?
* any baseDN for which data is imported will be cleared anyway (see getSuffix()),
* so if there is only one baseDN for this backend, then clear it now.
*/
}
{
{
{
}
else
{
}
}
return indexes;
}
/**
* Return the suffix instance in the specified map that matches the specified
* DN.
*
* @param dn
* The DN to search for.
* @param map
* The map to search.
* @return The suffix instance that matches the DN, or null if no match is
* found.
*/
{
{
{
}
}
return suffix;
}
/**
* Calculate buffer sizes and initialize properties based on memory.
*
* @throws InitializationException
* If a problem occurs during calculation.
*/
private void computeMemoryRequirements() throws InitializationException
{
// 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).
// We need caching when doing DN validation or rebuilding indexes.
{
// DN validation: calculate memory for DB cache, DN2ID temporary cache, and buffers.
{
}
{
}
else if (!clearedBackend)
{
// Appending to existing data so reserve extra memory for the DB cache
// since it will be needed for dn2id queries.
}
else
{
}
}
else
{
// No DN validation: calculate memory for DB cache and buffers.
// No need for DN2ID cache.
dnCacheSize = 0;
{
}
else if (usableMemory < MIN_DB_CACHE_MEMORY)
{
}
else
{
// not being queried.
}
}
final int oldThreadCount = threadCount;
{
while (true)
{
// Scratch writers allocate 4 buffers per index as well.
// We need (2 * bufferSize) to fit in an int for the insertByteStream
// and deleteByteStream constructors.
if (bufferSize > MAX_BUFFER_SIZE)
{
if (validateDNs)
{
// The buffers are big enough: the memory is best used for the DN2ID temp DB
if (!clearedBackend)
{
}
else
{
}
}
break;
}
else if (bufferSize > MIN_BUFFER_SIZE)
{
// This is acceptable.
break;
}
else if (threadCount > 1)
{
// Retry using less threads.
threadCount--;
}
else
{
// Not enough memory.
}
}
}
if (oldThreadCount != threadCount)
{
}
if (dnCacheSize > 0)
{
}
}
/**
* Calculates 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 void calculateAvailableMemory()
{
final long totalAvailableMemory;
if (DirectoryServer.isRunning())
{
}
else
{
}
// Now take into account various fudge factors.
int importMemPct = 90;
if (totalAvailableMemory <= SMALL_HEAP_SIZE)
{
// Be pessimistic when memory is low.
importMemPct -= 25;
}
if (rebuildManager != null)
{
// Rebuild seems to require more overhead.
importMemPct -= 15;
}
}
private void initializeIndexBuffers()
{
for (int i = 0; i < phaseOneBufferCount; i++)
{
}
}
private void initializeSuffixes(WriteableTransaction txn) throws ConfigException, DirectoryException
{
{
{
}
}
}
{
{
{
}
}
}
{
}
{
}
throws ConfigException, DirectoryException
{
{
return new Suffix(entryContainer);
}
{
// This entire base DN was explicitly excluded. Skip.
return null;
}
{
if (includeBranches.isEmpty())
{
// There are no branches in the explicitly defined include list under this base DN.
// Skip this base DN altogether.
return null;
}
// Remove any overlapping include branches.
while (includeBranchIterator.hasNext())
{
{
}
}
// Remove any exclude branches that are not are not under a include branch
// since they will be migrated as part of the existing entries
// outside of the include branches anyways.
while (excludeBranchIterator.hasNext())
{
{
}
}
if (excludeBranches.isEmpty()
{
// This entire base DN is explicitly included in the import with
// no exclude branches that we need to migrate.
// Just clear the entry container.
}
else
{
// Create a temp entry container
}
}
}
{
{
{
}
}
return results;
}
{
}
{
{
{
return false;
}
}
return true;
}
{
{
{
return true;
}
}
return false;
}
/**
* Rebuild the indexes using the specified root container.
*
* @throws ConfigException
* If a configuration error occurred.
* @throws InitializationException
* If an initialization error occurred.
* @throws StorageRuntimeException
* If the storage had an error.
* @throws InterruptedException
* If an interrupted error occurred.
* @throws ExecutionException
* If an execution error occurred.
*/
public void rebuildIndexes() throws ConfigException, InitializationException, StorageRuntimeException,
{
try
{
{
}
else
{
}
}
catch (Exception e)
{
logger.traceException(e);
}
}
private void clearDegradedState() throws Exception
{
{
{
}
});
}
private void rebuildIndexes0() throws Exception
{
{
{
}
});
{
{
}
});
}
{
try {
try
{
}
catch (IOException ioe)
{
}
{
{
setIndexesTrusted(txn, false);
}
});
if (validateDNs)
{
}
if (isCanceled)
{
throw new InterruptedException("Import processing canceled.");
}
if (isCanceled)
{
throw new InterruptedException("Import processing canceled.");
}
{
{
setIndexesTrusted(txn, true);
}
});
float rate = 0;
if (importTime > 0)
{
}
}
finally
{
if (validateDNs)
{
}
}
}
private void switchEntryContainers(WriteableTransaction txn) throws StorageRuntimeException, InitializationException
{
{
if (entryContainer != null)
{
replacement.lock();
}
}
}
private void setIndexesTrusted(WriteableTransaction txn, boolean trusted) throws StorageRuntimeException
{
try
{
{
}
}
catch (StorageRuntimeException ex)
{
}
}
/**
* Reads all entries from id2entry, and:
* <ol>
* <li>compute how the entry is indexed for each index</li>
* <li>store the result of indexing entries into in-memory index buffers</li>
* <li>each time an in-memory index buffer is filled, sort it and write it to scratch files.
* The scratch files will be read by phaseTwo to perform on-disk merge</li>
* </ol>
*/
private void importPhaseOne() throws Exception
{
{
for (int i = 0; i < threadCount; i++)
{
}
}
else
{
for (int i = 0; i < threadCount; i++)
{
}
}
// Try to clear as much memory as possible.
}
{
}
{
{
}
{
}
}
{
{
}
}
private void importPhaseTwo() throws Exception
{
try
{
}
finally
{
}
}
/**
* Performs on-disk merge by reading several scratch files at once
* and write their ordered content into the target indexes.
*/
private void processIndexFiles() throws Exception
{
{
return;
}
if (dbThreads < 4)
{
dbThreads = 4;
}
// Calculate memory / buffer counts.
int readAheadSize;
int buffers;
while (true)
{
buffers = 0;
for (int i = 0; i < limit; i++)
{
}
if (readAheadSize > bufferSize)
{
// Cache size is never larger than the buffer size.
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 - will need to do batching for the biggest indexes.
break;
}
}
// Ensure that there are minimum two threads available for parallel
// processing of smaller indexes.
// Start indexing tasks.
// Start DN processing first.
{
submitIndexDBWriteTasks(DNIndexMgrList, importer, dbService, permits, buffers, readAheadSize, futures);
submitIndexDBWriteTasks(indexMgrList, importer, dbService, permits, buffers, readAheadSize, futures);
}
finally
{
}
}
ExecutorService dbService, Semaphore permits, int buffers, int readAheadSize, List<Future<Void>> futures)
{
{
futures.add(dbService.submit(new IndexDBWriteTask(importer, indexMgr, permits, buffers, readAheadSize)));
}
}
private static <T> void getAll(List<Future<T>> futures) throws InterruptedException, ExecutionException
{
{
}
}
private void stopScratchFileWriters()
{
{
}
}
/** Task used to migrate excluded branch. */
private final class MigrateExcludedTask extends ImportTask
{
{
super(storage);
}
{
{
{
try
{
{
{
/*
* This is the base entry for a branch that was excluded in the
* import so we must migrate all entries in this branch over to
* the new entry container.
*/
while (success
&& !importCfg.isCancelled()
&& !isCanceled)
{
}
}
}
}
catch (Exception e)
{
isCanceled = true;
throw e;
}
finally
{
}
}
}
}
}
/** Task to migrate existing entries. */
private final class MigrateExistingTask extends ImportTask
{
{
super(storage);
}
{
{
{
try
{
while (success
&& !importCfg.isCancelled()
&& !isCanceled)
{
{
}
else
{
/*
* This is the base entry for a branch that will be included
* in the import so we do not want to copy the branch to the
* new entry container.
*/
/*
* Advance the cursor to next entry at the same level in the DIT
* skipping all the entries in this branch.
*/
}
}
}
catch (Exception e)
{
isCanceled = true;
throw e;
}
finally
{
}
}
}
}
{
{
{
}
}
return includeBranches;
}
}
private class AppendReplaceTask extends ImportTask
{
{
super(storage);
}
{
try
{
{
{
return;
}
}
}
catch (Exception e)
{
isCanceled = true;
throw e;
}
}
{
{
{
return;
}
}
else
{
}
{
}
else
{
}
}
{
{
}
}
{
{
{
}
}
{
}
}
}
/**
* This task performs phase reading and processing of the entries read from
* the LDIF file(s). This task is used if the append flag wasn't specified.
*/
{
{
}
/** {@inheritDoc} */
{
{
{
}
});
return null;
}
{
try
{
{
{
return;
}
}
}
catch (Exception e)
{
isCanceled = true;
throw e;
}
}
{
{
return;
}
}
/**
* Examine the DN for duplicates and missing parents.
*
* @return true if the import operation can proceed with the provided entry, false otherwise
*/
{
//Perform parent checking.
{
return false;
}
{
return false;
}
return true;
}
{
//If the backend was not cleared, then first check dn2id
//for DNs that might not exist in the DN cache.
{
return false;
}
}
{
{
{
}
}
}
{
{
}
}
throws DirectoryException
{
{
}
}
{
{
}
}
{
{
}
}
{
{
}
}
throws InterruptedException
{
if (indexBuffer == null)
{
}
{
// complete the current buffer...
// ... and get a new one
}
return indexID;
}
{
if (size > bufferSize)
{
}
else
{
if (indexBuffer == null)
{
throw new InterruptedException("Index buffer processing error.");
}
}
if (indexBuffer.isPoison())
{
throw new InterruptedException("Cancel processing received.");
}
return indexBuffer;
}
throws InterruptedException
{
}
}
/**
* This task reads sorted records from the temporary index scratch files,
* processes the records and writes the results to the index tree. The DN
* index is treated differently then non-DN indexes.
*/
{
private final IndexManager indexMgr;
private final int cacheSize;
/** indexID => DNState map */
private final int maxPermits;
private long lastBytesRead;
private RandomAccessFile bufferFile;
private DataInputStream bufferIndexFile;
private int remainingBuffers;
private volatile int totalBatches;
private int nextBufferID;
private int ownedPermits;
private volatile boolean isRunning;
/**
* Creates a new index DB writer.
*
* @param importer
* The importer
* @param indexMgr
* The index manager.
* @param permits
* The semaphore used for restricting the number of buffer allocations.
* @param maxPermits
* The maximum number of buffers which can be allocated.
* @param cacheSize
* The buffer cache size.
*/
public IndexDBWriteTask(org.opends.server.backends.pluggable.spi.Importer importer, IndexManager indexMgr,
{
this.maxPermits = maxPermits;
}
/**
* Initializes this task.
*
* @throws IOException
* If an IO error occurred.
*/
public void beginWriteTask() throws IOException
{
indexMgr.getBufferIndexFile())));
nextBufferID = 0;
ownedPermits = 0;
logger.info(NOTE_IMPORT_LDIF_INDEX_STARTED, indexMgr.getBufferFileName(), remainingBuffers, totalBatches);
indexMgr.setIndexDBWriteTask(this);
isRunning = true;
}
/**
* Returns the next batch of buffers to be processed, blocking until enough
* buffer permits are available.
*
* @return The next batch of buffers, or {@code null} if there are no more
* buffers to be processed.
* @throws Exception
* If an exception occurred.
*/
{
// First release any previously acquired permits.
if (ownedPermits > 0)
{
ownedPermits = 0;
}
// Block until we can either get enough permits for all buffers, or the
// maximum number of permits.
if (permitRequest == 0)
{
// No more work to do.
return null;
}
// Update counters.
// Create all the index buffers for the next batch.
for (int i = 0; i < permitRequest; i++)
{
}
return buffers;
}
/** Finishes this task. */
{
isRunning = false;
// First release any previously acquired permits.
if (ownedPermits > 0)
{
ownedPermits = 0;
}
try
{
{
{
}
if (!isCanceled)
{
}
}
else
{
if (!isCanceled)
{
}
}
}
finally
{
}
}
/**
* Print out progress stats.
*
* @param deltaTime
* The time since the last update.
*/
public void printStats(long deltaTime)
{
if (isRunning)
{
final int bytesReadPercent =
// Kilo and milli approximately cancel out.
}
}
/** {@inheritDoc} */
{
return null;
}
{
if (isCanceled)
{
return;
}
try
{
{
if (isCanceled)
{
return;
}
{
{
if (previousRecord != null)
{
}
// this is a new record
}
// merge all entryIds into the idSets
if (b.hasMoreData())
{
b.fetchNextRecord();
}
}
if (previousRecord != null)
{
}
}
}
catch (Exception e)
{
throw e;
}
finally
{
}
}
{
{
}
}
private void addToDB(org.opends.server.backends.pluggable.spi.Importer importer, int indexID, ImportIDSet insertSet,
{
{
}
else
{
{
}
{
}
}
}
private void addDN2ID(org.opends.server.backends.pluggable.spi.Importer importer, int indexID, ImportIDSet idSet)
throws DirectoryException
{
{
}
{
}
}
private void addBytesRead(int bytesRead)
{
}
/**
* This class is used to by a index DB merge thread performing DN processing
* to keep track of the state of individual DN2ID index processing.
*/
private final class DNState
{
private final EntryContainer entryContainer;
private ByteSequence parentDN;
private long totalNbEntries;
{
this.entryContainer = entryContainer;
}
{
if (parentIndex < 0)
{
// This is the root or base DN
return null;
}
}
/** Why do we still need this if we are checking parents in the first phase? */
throws StorageRuntimeException
{
if (bypassCacheForAppendMode())
{
// If null is returned then this is a suffix DN.
{
{
// We have a missing parent. Maybe parent checking was turned off?
// Just ignore.
return false;
}
}
}
else if (parentIDMap.isEmpty())
{
return true;
}
{
return true;
}
{
return true;
}
{
{
}
}
else
{
// We have a missing parent. Maybe parent checking was turned off?
// Just ignore.
return false;
}
return true;
}
private AtomicLong getId2childrenCounter()
{
{
counter = new AtomicLong();
}
return counter;
}
/**
* For append data, bypass the {@link #parentIDMap} cache, and lookup the parent DN in the
* DN2ID index.
*/
private boolean bypassCacheForAppendMode()
{
}
private EntryID get(org.opends.server.backends.pluggable.spi.Importer importer, TreeName dn2id, ByteSequence dn)
throws StorageRuntimeException
{
}
throws DirectoryException
{
{
}
}
{
{
}
}
{
{
}
}
{
}
}
}
/**
* This task writes the temporary scratch index files using the sorted buffers
* read from a blocking queue private to each index.
*/
{
private static final int DRAIN_TO = 3;
private final IndexManager indexMgr;
/** Stream where to output insert ImportIDSet data. */
/** Stream where to output delete ImportIDSet data. */
private final DataOutputStream bufferStream;
private final DataOutputStream bufferIndexStream;
private int insertKeyCount, deleteKeyCount;
private int bufferCount;
private boolean poisonSeen;
{
}
{
return new DataOutputStream(new BufferedOutputStream(new FileOutputStream(file), READER_WRITER_BUFFER_SIZE));
}
/** {@inheritDoc} */
{
long offset = 0;
try
{
while (true)
{
long beginOffset = offset;
long bufferLen;
{
l.add(indexBuffer);
bufferLen = writeIndexBuffers(l);
for (IndexOutputBuffer id : l)
{
if (!id.isDiscarded())
{
}
}
l.clear();
}
else
{
if (indexBuffer.isPoison())
{
break;
}
if (!indexBuffer.isDiscarded())
{
indexBuffer.reset();
}
}
// Write buffer index information.
bufferCount++;
if (poisonSeen)
{
break;
}
}
return null;
}
catch (IOException e)
{
logger.error(ERR_IMPORT_LDIF_INDEX_FILEWRITER_ERR, indexMgr.getBufferFile().getAbsolutePath(), e.getMessage());
isCanceled = true;
throw e;
}
finally
{
}
}
{
long bufferLen = 0;
for (int i = 0; i < numberKeys; i++)
{
if (i == 0)
{
// first record, initialize all
resetStreams();
}
else if (!indexBuffer.sameKeyAndIndexID(i))
{
// this is a new record, save previous record ...
// ... and reinitialize all
resetStreams();
}
}
if (numberKeys > 0)
{
// save the last record
}
return bufferLen;
}
{
resetStreams();
long bufferID = 0;
long bufferLen = 0;
for (IndexOutputBuffer b : buffers)
{
if (b.isPoison())
{
poisonSeen = true;
}
else
{
b.setPosition(0);
b.setBufferID(bufferID++);
indexSortedSet.add(b);
}
}
while (!indexSortedSet.isEmpty())
{
{
if (previousRecord != null)
{
resetStreams();
}
// this is a new record
previousRecord = b.currentRecord();
}
appendNextEntryIDToStream(b, b.getPosition());
if (b.hasMoreData())
{
b.nextRecord();
indexSortedSet.add(b);
}
}
if (previousRecord != null)
{
}
return bufferLen;
}
private void resetStreams()
{
insertKeyCount = 0;
deleteKeyCount = 0;
}
{
{
{
}
// else do not bother appending, this value will not be read.
// instead, a special value will be written to show the index entry limit is exceeded
}
else
{
}
}
private int writeByteStreams() throws IOException
{
{
// special handling when index entry limit has been exceeded
insertKeyCount = 1;
}
int insertSize = INT_SIZE;
{
}
int deleteSize = INT_SIZE;
{
}
}
{
return 2 * INT_SIZE;
}
{
int bodySize = writeByteStreams();
}
/** {@inheritDoc} */
{
return getClass().getSimpleName() + "(" + indexMgr.getBufferFileName() + ": " + indexMgr.getBufferFile() + ")";
}
}
/**
* This task main function is to sort the index buffers given to it from the
* import tasks reading the LDIF file. It will also create a index file writer
* task and corresponding queue if needed. The sorted index buffers are put on
* the index file writer queues for writing to a temporary file.
*/
{
private final IndexOutputBuffer indexBuffer;
{
this.indexBuffer = indexBuffer;
}
/** {@inheritDoc} */
{
|| isCanceled)
{
isCanceled = true;
return null;
}
indexBuffer.sort();
{
}
return null;
}
{
synchronized (synObj)
{
{
return;
}
if (isDN2ID)
{
}
else
{
}
}
}
}
/**
* The index manager class has several functions:
* <ol>
* <li>It is used to carry information about index processing created in phase one to phase two</li>
* <li>It collects statistics about phase two processing for each index</li>
* <li>It manages opening and closing the scratch index files</li>
* </ol>
*/
{
private final File bufferFile;
private final String bufferFileName;
private final File bufferIndexFile;
private final boolean isDN2ID;
private final int indexEntryLimit;
private int numberOfBuffers;
private long bufferFileSize;
private long totalDNs;
private volatile IndexDBWriteTask writer;
{
this.bufferFileName = fileName;
}
{
}
private File getBufferFile()
{
return bufferFile;
}
private long getBufferFileSize()
{
return bufferFileSize;
}
private File getBufferIndexFile()
{
return bufferIndexFile;
}
{
this.numberOfBuffers = numberOfBuffers;
this.bufferFileSize = bufferFileSize;
}
/**
* Updates the bytes read counter.
*
* @param bytesRead
* The number of bytes read.
*/
void addBytesRead(int bytesRead)
{
{
}
}
private void addTotDNCount(int delta)
{
}
private long getDNCount()
{
return totalDNs;
}
private boolean isDN2ID()
{
return isDN2ID;
}
private void printStats(long deltaTime)
{
{
}
}
/**
* Returns the file name associated with this index manager.
*
* @return The file name associated with this index manager.
*/
{
return bufferFileName;
}
private int getIndexEntryLimit()
{
return indexEntryLimit;
}
/** {@inheritDoc} */
{
}
private int getNumberOfBuffers()
{
return numberOfBuffers;
}
/** {@inheritDoc} */
{
}
}
/** The rebuild index manager handles all rebuild index related processing. */
private class RebuildIndexManager extends ImportTask
{
/** Rebuild index configuration. */
private final RebuildConfig rebuildConfig;
/** Backend configuration. */
private final PluggableBackendCfg cfg;
/** Map of index keys to indexes to rebuild. */
/** List of VLV indexes to rebuild. */
private boolean reBuildDn2id;
private boolean rebuildDn2uri;
/** The suffix instance. */
/** The entry container. */
private EntryContainer entryContainer;
/** Total entries to be processed. */
private long totalEntries;
/** Total entries processed. */
{
super(storage);
this.rebuildConfig = rebuildConfig;
}
{
}
{
switch (rebuildConfig.getRebuildMode())
{
case ALL:
break;
case DEGRADED:
break;
default:
&& logger.isInfoEnabled())
{
}
break;
}
}
void printStopMessage(long rebuildStartTime)
{
float rate = 0;
if (totalTime > 0)
{
}
if (!rebuildConfig.isClearDegradedState())
{
}
}
{
try
{
{
if (isCanceled)
{
return;
}
}
}
catch (Exception e)
{
logger.traceException(e);
isCanceled = true;
throw e;
}
finally
{
}
}
{
}
{
setRebuildListIndexesTrusted(txn, false);
}
private void throwIfCancelled() throws InterruptedException
{
if (isCanceled)
{
throw new InterruptedException("Rebuild Index canceled.");
}
}
{
setRebuildListIndexesTrusted(txn, true);
}
{
// Depends on rebuild mode, (re)building indexes' lists.
switch (mode)
{
case ALL:
reBuildDn2id = true;
rebuildDn2uri = true;
rebuildIndexMap(txn, false);
break;
case DEGRADED:
rebuildIndexMap(txn, true);
if (vlvIndexes.isEmpty())
{
}
break;
case USER_DEFINED:
// false may be required if the user wants to rebuild specific index.
rebuildIndexMap(txn, false);
break;
default:
break;
}
}
{
{
if (mustRebuild(attributeType))
{
}
}
}
{
switch (rebuildConfig.getRebuildMode())
{
case ALL:
case DEGRADED:
// Get all existing indexes
return true;
case USER_DEFINED:
// Get the user selected indexes
{
{
return true;
}
}
return false;
default:
return false;
}
}
private void rebuildAttributeIndexes(WriteableTransaction txn, AttributeIndex attrIndex, AttributeType attrType,
boolean onlyDegraded) throws StorageRuntimeException
{
{
{
}
}
}
{
if (rebuildDn2uri)
{
}
if (reBuildDn2id)
{
}
{
{
}
}
{
{
}
}
}
private void setRebuildListIndexesTrusted(WriteableTransaction txn, boolean trusted) throws StorageRuntimeException
{
try
{
{
}
{
}
}
catch (StorageRuntimeException ex)
{
}
}
/** @see Importer#importPhaseOne(WriteableTransaction) */
{
for (int i = 0; i < threadCount; i++)
{
}
// Try to clear as much memory as possible.
}
private void rebuildIndexesPhaseTwo() throws Exception
{
try
{
}
finally
{
}
}
{
return timer;
}
private int getIndexCount() throws ConfigException, StorageRuntimeException, InitializationException
{
switch (rebuildConfig.getRebuildMode())
{
case ALL:
return getTotalIndexCount(cfg);
case DEGRADED:
// FIXME: since the storage is not opened we cannot determine which indexes are degraded.
// As a workaround, be conservative and assume all indexes need rebuilding.
return getTotalIndexCount(cfg);
default:
return getRebuildListIndexCount(cfg);
}
}
{
if (rebuildList.isEmpty())
{
return 0;
}
int indexCount = 0;
{
{
indexCount += 3;
}
{
indexCount++;
}
{
{
}
indexCount++;
}
{
throw attributeIndexNotConfigured(index);
}
else
{
{
throw attributeIndexNotConfigured(index);
}
{
throw attributeIndexNotConfigured(index);
}
{
{
{
indexCount++;
}
else
{
throw attributeIndexNotConfigured(index);
}
}
else // attrIndexParts.length == 3
{
{
throw attributeIndexNotConfigured(index);
}
indexCount++;
}
}
else
{
boolean found = false;
{
{
found = true;
}
}
if (!found)
{
throw attributeIndexNotConfigured(index);
}
}
}
}
return indexCount;
}
{
}
private boolean findExtensibleMatchingRule(PluggableBackendCfg cfg, String indexExRuleName) throws ConfigException
{
{
{
{
{
return true;
}
}
}
}
return false;
}
{
int result = 0;
{
{
result++;
}
}
return result;
}
{
int result = 0;
{
boolean shared = false;
{
{
result++;
}
else if (!shared)
{
shared = true;
result++;
}
}
}
return result;
}
{
if (reBuildDn2id)
{
}
if (rebuildDn2uri)
{
}
}
{
{
}
}
{
{
{
}
}
}
/**
* Return the number of entries processed by the rebuild manager.
*
* @return The number of entries processed.
*/
long getEntriesProcessed()
{
return this.entriesProcessed.get();
}
/**
* Return the total number of entries to process by the rebuild manager.
*
* @return The total number for entries to process.
*/
long getTotalEntries()
{
return this.totalEntries;
}
}
/**
* This class reports progress of rebuild index processing at fixed intervals.
*/
private class RebuildFirstPhaseProgressTask extends TimerTask
{
/**
* The number of records that had been processed at the time of the previous
* progress report.
*/
private long previousProcessed;
/** The time in milliseconds of the previous progress report. */
private long previousTime;
/**
* Create a new rebuild index progress task.
*
* @throws StorageRuntimeException
* If an error occurred while accessing the storage.
*/
public RebuildFirstPhaseProgressTask() throws StorageRuntimeException
{
}
/**
* The action to be performed by this timer task.
*/
public void run()
{
if (deltaTime == 0)
{
return;
}
float completed = 0;
{
}
logger.info(NOTE_REBUILD_PROGRESS_REPORT, completed, entriesProcessed, rebuildManager.getTotalEntries(), rate);
}
}
/**
* This class reports progress of first phase of import processing at fixed
* intervals.
*/
private final class FirstPhaseProgressTask extends TimerTask
{
/**
* The number of entries that had been read at the time of the previous
* progress report.
*/
private long previousCount;
/** The time in milliseconds of the previous progress report. */
private long previousTime;
/** Create a new import progress task. */
public FirstPhaseProgressTask()
{
}
/** The action to be performed by this timer task. */
public void run()
{
if (deltaTime == 0)
{
return;
}
}
}
/**
* This class reports progress of the second phase of import processing at
* fixed intervals.
*/
private class SecondPhaseProgressTask extends TimerTask
{
/** The time in milliseconds of the previous progress report. */
private long previousTime;
/** Create a new import progress task. */
public SecondPhaseProgressTask()
{
}
/** The action to be performed by this timer task. */
public void run()
{
if (deltaTime == 0)
{
return;
}
//Do DN index managers first.
{
}
//Do non-DN index managers.
{
}
}
}
/**
* This class is used as an index key for hash maps that need to process multiple suffix index
* cn.equality, sn.equality,...).
*/
static class IndexKey
{
private final AttributeType attributeType;
private final int entryLimit;
/**
* Create index key instance using the specified attribute type, index ID and index entry
* limit.
*
* @param attributeType
* The attribute type.
* @param indexID
* The index ID taken from the matching rule's indexer.
* @param entryLimit
* The entry limit for the index.
*/
{
this.attributeType = attributeType;
this.entryLimit = entryLimit;
}
/**
* An equals method that uses both the attribute type and the index ID. Only returns
* {@code true} if the attribute type and index ID are equal.
*
* @param obj
* the object to compare.
* @return {@code true} if the objects are equal, or {@code false} if they are not.
*/
{
{
{
return true;
}
}
return false;
}
/**
* A hash code method that adds the hash codes of the attribute type and index ID and returns
* that value.
*
* @return The combined hash values of attribute type hash code and the index ID hash code.
*/
public int hashCode()
{
}
/**
* Return the attribute type.
*
* @return The attribute type.
*/
private AttributeType getAttributeType()
{
return attributeType;
}
/**
* Return the index ID.
*
* @return The index ID.
*/
private String getIndexID()
{
return indexID;
}
/**
* Return the index key name, which is the attribute type primary name, a period, and the index
* ID name. Used for building file names and progress output.
*
* @return The index key name.
*/
{
}
/**
* Return the entry limit associated with the index.
*
* @return The entry limit.
*/
private int getEntryLimit()
{
return entryLimit;
}
/** {@inheritDoc} */
{
return getClass().getSimpleName()
+ ", entryLimit=" + entryLimit
+ ")";
}
}
/**
* This interface is used by those suffix instance to do parental checking of the DN cache.
* <p>
* It will be shared when multiple suffixes are being processed.
*/
{
/**
* Insert the specified DN into the DN cache. It will return {@code true} if the DN does not
* already exist in the cache and was inserted, or {@code false} if the DN exists already in the
* cache.
*
* @param dn
* The DN to insert in the cache.
* @return {@code true} if the DN was inserted in the cache, or {@code false} if the DN exists
* in the cache already and could not be inserted.
* @throws StorageRuntimeException
* If an error occurs accessing the storage.
*/
/**
* Returns whether the specified DN is contained in the DN cache.
*
* @param dn
* The DN to check the presence of.
* @return {@code true} if the cache contains the DN, or {@code false} if it
* is not.
* @throws StorageRuntimeException
* If an error occurs reading the storage.
*/
/**
* Shuts the DN cache down.
*
* @throws StorageRuntimeException
* If error occurs.
*/
void close();
}
/** Invocation handler for the {@link PluggableBackendCfg} proxy. */
private static final class BackendCfgHandler implements InvocationHandler
{
{
this.returnValues = returnValues;
}
{
if ((methodName.startsWith("add") || methodName.startsWith("remove")) && methodName.endsWith("ChangeListener"))
{
// ignore calls to (add|remove)*ChangeListener() methods
return null;
}
if (returnValue != null)
{
return returnValue;
}
throw new IllegalArgumentException("Unhandled method call on proxy ("
+ BackendCfgHandler.class.getSimpleName()
+ ") for method (" + method
}
}
/**
* Used to check DN's when DN validation is performed during phase one processing.
* It is deleted after phase one processing.
*/
private final class DNCacheImpl implements DNCache
{
{
try
{
{
{
}
});
}
catch (Exception e)
{
throw new StorageRuntimeException(e);
}
}
{
getClass().getClassLoader(),
new Class<?>[] { PersistitBackendCfg.class },
new BackendCfgHandler(returnValues));
}
private static final long FNV_INIT = 0xcbf29ce484222325L;
private static final long FNV_PRIME = 0x100000001b3L;
/** Hash the DN bytes. Uses the FNV-1a hash. */
{
for (int i = 0; i < b.length(); i++)
{
}
}
public void close() throws StorageRuntimeException
{
try
{
}
finally
{
}
}
{
// Use a compact representation for key
// and a reversible representation for value
}
private boolean insert(final ByteString key, final ByteStringBuilder dn) throws StorageRuntimeException
{
try
{
{
{
{
{
{
// no change
return existingDns;
}
else if (existingDns != null)
{
}
else
{
return singletonList(dn);
}
}
/** Add the DN to the DNs because of a hash collision. */
{
final ByteStringBuilder builder = new ByteStringBuilder(dnList.length() + INT_SIZE + dntoAdd.length());
return builder;
}
/** Create a list of dn made of one element. */
{
return singleton;
}
}));
}
});
return updateResult.get();
}
catch (StorageRuntimeException e)
{
throw e;
}
catch (Exception e)
{
throw new StorageRuntimeException(e);
}
}
/** Return true if the specified DN is in the DNs saved as a result of hash collisions. */
{
{
// TODO JNR remove call to toByteArray() on next line?
int previousPos = 0;
{
{
return true;
}
}
}
return false;
}
{
try
{
{
{
if (existingDns != null)
{
}
return false;
}
});
}
catch (StorageRuntimeException e)
{
throw e;
}
catch (Exception e)
{
throw new StorageRuntimeException(e);
}
}
}
}