Importer.java revision 7ccaa46b4a749896b2daabda390d8ddd3ae7743f
/*
* 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
{
private static final int TIMER_INTERVAL = 10000;
private static final int KB = 1024;
/** Defaults for DB cache. */
/**
* Defaults for LDIF reader buffers, min memory required to import and default
* size for byte buffers.
*/
private static final int MIN_DB_CACHE_MEMORY = MAX_DB_CACHE_SIZE
private static final int BYTE_BUFFER_CAPACITY = 128;
/** 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;
/** The dn2id "index ID". */
/** Phase one buffer count. */
/** Phase one imported entries count. */
/** Phase one buffer size in bytes. */
private int bufferSize;
/** Temp scratch directory. */
/** Index count. */
private final int indexCount;
/** Thread count. */
private int threadCount;
/** Set to true when validation is skipped. */
private final boolean skipDNValidation;
/** Temporary environment used when DN validation is done in first phase. */
/** Root container. */
private RootContainer rootContainer;
/** Import configuration. */
private final LDIFImportConfig importConfiguration;
private final ServerContext serverContext;
/** LDIF reader. */
private ImportLDIFReader reader;
/** Migrated entry count. */
private int migratedCount;
/** Size in bytes of temporary env. */
private long tmpEnvCacheSize;
/** 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. */
new LinkedBlockingQueue<IndexOutputBuffer>();
/**
* Map of index keys to index buffers. Used to allocate sorted index buffers
* to a index writer thread.
*/
/** 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 database containers. */
private final ConcurrentHashMap<Integer, Index> indexIDToIndexMap = new ConcurrentHashMap<Integer, Index>();
/** Map of indexIDs to entry containers. */
/** 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;
private volatile boolean isPhaseOneDone;
/** Number of phase one buffers. */
private int phaseOneBufferCount;
static
{
{
}
}
/**
* 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.
* @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.
*/
{
this.importConfiguration = null;
this.serverContext = serverContext;
this.threadCount = 1;
this.clearedBackend = false;
this.scratchFileWriterList =
{
}
this.skipDNValidation = true;
}
/**
* Create a new import job with the specified ldif import config.
*
* @param importConfiguration
* The LDIF import configuration.
* @param backendCfg
* The local DB back-end 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.
*/
Importer(LDIFImportConfig importConfiguration, PluggableBackendCfg backendCfg, ServerContext serverContext)
{
this.rebuildManager = null;
this.serverContext = serverContext;
{
}
else
{
}
// Determine the number of indexes.
this.scratchFileWriterList =
{
}
// Set up temporary environment.
if (!skipDNValidation)
{
}
else
{
}
}
/**
* 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#getSuffix(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.
*/
}
{
if (tmpDirectory != null)
{
}
else
{
}
}
throws ConfigException
{
{
{
}
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 JEB properties based on memory.
*
* @throws InitializationException
* If a problem occurs during calculation.
*/
private void initializeDBEnv() 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.
{
// No 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.
tmpEnvCacheSize = 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 (!skipDNValidation)
{
// 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.
throw new InitializationException(message);
}
}
}
if (oldThreadCount != threadCount)
{
}
if (tmpEnvCacheSize > 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++)
{
}
}
{
{
{
}
}
}
/**
* Mainly used to support multiple suffixes. Each index in each suffix gets an
* unique ID to identify which DB it needs to go to in phase two processing.
*/
{
{
{
}
}
}
{
{
}
}
{
}
throws ConfigException
{
&& !importConfiguration.clearBackend())
{
{
{
// 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 all together.
*/
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
}
}
}
}
private EntryContainer createEntryContainer(WriteableTransaction txn, DN baseDN) throws ConfigException
{
try
{
}
catch (DirectoryException e)
{
throw new ConfigException(e.getMessageObject());
}
}
{
}
{
{
{
return false;
}
}
return true;
}
{
{
{
return true;
}
}
return false;
}
/**
* Rebuild the indexes using the specified root container.
*
* @param rootContainer
* The root container to rebuild indexes in.
* @throws ConfigException
* If a configuration error occurred.
* @throws InitializationException
* If an initialization error occurred.
* @throws StorageRuntimeException
* If the JEB database had an error.
* @throws InterruptedException
* If an interrupted error occurred.
* @throws ExecutionException
* If an execution error occurred.
*/
{
this.rootContainer = rootContainer;
try
{
{
}
else
{
}
}
catch (Exception e)
{
logger.traceException(e);
}
}
private void clearDegradedState() throws Exception
{
{
{
}
});
}
private void rebuildIndexes() throws Exception
{
{
{
}
});
{
{
}
});
}
/**
* Import a LDIF using the specified root container.
*
* @param rootContainer
* The root container to use during the import.
* @return A LDIF result.
* @throws Exception
* If the import failed
*/
{
this.rootContainer = rootContainer;
try {
try
{
}
catch (IOException ioe)
{
}
{
{
setIndexesTrusted(txn, false);
}
});
isPhaseOneDone = true;
if (!skipDNValidation)
{
}
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 (!skipDNValidation)
{
try
{
}
{
// Do nothing.
}
}
}
}
{
{
{
if (f.isDirectory())
{
recursiveDelete(f);
}
f.delete();
}
}
}
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)
{
throw new StorageRuntimeException(NOTE_JEB_IMPORT_LDIF_TRUSTED_FAILED.get(ex.getMessage()).toString());
}
}
/**
* 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.
}
{
}
{
{
}
{
}
}
{
{
}
}
{
try
{
}
finally
{
}
}
/**
* Performs on-disk merge by reading several scratch files at once
* and write their ordered content into the target indexes.
*/
{
{
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.
}
{
{
// avoid threading issues by allocating one writeable storage per thread
// DB transactions are generally tied to a single thread
futures.add(dbService.submit(new IndexDBWriteTask(indexMgr, txn, permits, buffers, readAheadSize)));
}
}
{
{
}
}
private void stopScratchFileWriters()
{
{
}
}
/** Task used to migrate excluded branch. */
private final class MigrateExcludedTask extends ImportTask
{
{
super(txn);
}
/** {@inheritDoc} */
{
{
{
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
&& !isCanceled)
{
}
}
}
}
catch (Exception e)
{
isCanceled = true;
throw e;
}
finally
{
}
}
}
return null;
}
}
/** Task to migrate existing entries. */
private final class MigrateExistingTask extends ImportTask
{
{
super(txn);
}
/** {@inheritDoc} */
{
{
{
try
{
while (success
&& !isCanceled)
{
{
}
else
{
// This is the base entry for a branch that will be included
// in the import so we don't 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. Set the next
* starting value to a value of equal length but slightly
* greater than the previous DN. Since keys are compared in
* reverse order we must set the first byte (the comma). No
* possibility of overflow here.
*/
}
}
}
catch (Exception e)
{
isCanceled = true;
throw e;
}
finally
{
}
}
}
return null;
}
{
{
{
}
}
return includeBranches;
}
}
/**
*/
private class AppendReplaceTask extends ImportTask
{
{
super(txn);
}
/** {@inheritDoc} */
{
try
{
while (true)
{
{
return null;
}
{
break;
}
}
return null;
}
catch (Exception e)
{
isCanceled = true;
throw e;
}
finally
{
}
}
{
{
}
{
{
return;
}
}
else
{
}
{
}
else
{
}
}
{
{
}
}
void processAttribute(MatchingRuleIndex index, Entry entry, EntryID entryID, IndexingOptions options,
{
{
{
}
}
{
}
}
}
/**
* 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.
*/
{
private final Map<IndexKey, IndexOutputBuffer> indexBufferMap = new HashMap<IndexKey, IndexOutputBuffer>();
{
}
/** {@inheritDoc} */
{
try
{
while (true)
{
{
return null;
}
{
break;
}
}
return null;
}
catch (Exception e)
{
isCanceled = true;
throw e;
}
finally
{
}
}
{
{
return;
}
}
/** Examine the DN for duplicates and missing parents. */
{
//Perform parent checking.
{
return false;
}
//If the backend was not cleared, then the dn2id needs to checked first
//for DNs that might not exist in the DN cache. If the DN is not in
//the suffixes dn2id DB, then the dn cache is used.
if (!clearedBackend)
{
{
return false;
}
}
{
return false;
}
return true;
}
{
{
{
}
}
}
{
{
}
}
{
{
}
}
{
{
}
}
void processAttribute(MatchingRuleIndex index, Entry entry, EntryID entryID, IndexingOptions options,
{
{
}
}
{
{
}
}
{
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
{
}
{
{
}
else
{
}
}
}
/**
* This task reads sorted records from the temporary index scratch files,
* processes the records and writes the results to the index database. 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;
private final WriteableTransaction txn;
/**
* Creates a new index DB writer.
*
* @param indexMgr
* The index manager.
* @param txn
* The database transaction
* @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(IndexManager indexMgr, WriteableTransaction txn, Semaphore permits, int maxPermits,
int cacheSize)
{
this.maxPermits = maxPermits;
}
/**
* Initializes this task.
*
* @throws IOException
* If an IO error occurred.
*/
public void beginWriteTask() throws IOException
{
indexMgr.getBufferIndexFile())));
nextBufferID = 0;
ownedPermits = 0;
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.
*/
public void endWriteTask()
{
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} */
{
if (isCanceled)
{
return null;
}
try
{
{
if (isCanceled)
{
return null;
}
{
{
if (previousRecord != null)
{
}
// this is a new record, reinitialize all
int indexID = b.getIndexID();
}
// merge all entryIds into the idSets
if (b.hasMoreData())
{
b.fetchNextRecord();
}
}
if (previousRecord != null)
{
}
}
return null;
}
catch (Exception e)
{
throw e;
}
finally
{
endWriteTask();
}
}
{
{
}
}
private void addToDB(int indexID, ImportIDSet insertSet, ImportIDSet deleteSet) throws DirectoryException
{
{
}
else
{
{
}
{
}
}
}
{
{
}
{
}
}
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.
*/
final class DNState
{
private final EntryContainer entryContainer;
private final int childLimit, subTreeLimit;
private final boolean childDoCount, subTreeDoCount;
private ByteSequence parentDN;
{
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? */
private boolean checkParent(ReadableTransaction txn, ImportIDSet idSet) throws StorageRuntimeException
{
//Bypass the cache for append data, lookup the parent in DN2ID and return.
if (importConfiguration != null
{
//If null is returned than this is a suffix DN.
{
{
}
else
{
// 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 ImportIDSet getId2childtreeImportIDSet()
{
{
}
return idSet;
}
private EntryID getParentID(ReadableTransaction txn, ByteSequence dn) throws StorageRuntimeException
{
// Bypass the cache for append data, lookup the parent DN in the DN2ID db
{
}
}
{
{
}
// TODO:
// Instead of doing this,
// we can just walk to parent cache if available
{
{
}
// else we have a missing parent. Maybe parent checking was turned off?
// Just ignore.
}
{
}
}
{
{
}
return idSet;
}
{
{
}
}
public void flush()
{
}
{
{
}
if (clearMap)
{
}
}
}
}
/**
* This task writes the temporary scratch index files using the sorted buffers
* read from a blocking queue private to each index.
*/
{
private 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)
{
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;
/** Local DB backend configuration. */
private final PluggableBackendCfg cfg;
/** Map of index keys to indexes. */
/** List of VLV indexes. */
/** The DN2ID index. */
/** The DN2URI index. */
/** Total entries to be processed. */
private long totalEntries;
/** Total entries processed. */
/** The suffix instance. */
/** The entry container. */
private EntryContainer entryContainer;
/**
* Create an instance of the rebuild index manager using the specified
* parameters.
*
* @param rebuildConfig
* The rebuild configuration to use.
* @param cfg
* The local DB configuration to use.
*/
{
super(null);
this.rebuildConfig = rebuildConfig;
}
/**
* Initialize a rebuild index manager.
*
* @throws ConfigException
* If an configuration error occurred.
* @throws InitializationException
* If an initialization error occurred.
*/
{
{
throw new InitializationException(
}
}
/**
* Print start message.
*/
{
switch (rebuildConfig.getRebuildMode())
{
case ALL:
break;
case DEGRADED:
break;
default:
&& logger.isInfoEnabled())
{
}
break;
}
}
/**
* Print stop message.
*
* @param startTime
* The time the rebuild started.
*/
public void printStopMessage(long startTime)
{
float rate = 0;
if (totalTime > 0)
{
}
if (!rebuildConfig.isClearDegradedState())
{
}
}
/** {@inheritDoc} */
{
try
{
{
if (isCanceled)
{
return null;
}
}
return null;
}
catch (Exception e)
{
logger.traceException(e);
isCanceled = true;
throw e;
}
finally
{
}
}
{
}
{
setRebuildListIndexesTrusted(txn, false);
clearIndexes(txn, true);
}
private void throwIfCancelled() throws InterruptedException
{
if (isCanceled)
{
throw new InterruptedException("Rebuild Index canceled.");
}
}
{
setRebuildListIndexesTrusted(txn, true);
}
@SuppressWarnings("fallthrough")
{
// Depends on rebuild mode, (re)building indexes' lists.
switch (mode)
{
case ALL:
rebuildIndexMap(txn, false);
// falls through
case DEGRADED:
{
}
{
}
{
}
{
}
break;
case USER_DEFINED:
// false may be required if the user wants to rebuild specific index.
rebuildIndexMap(txn, false);
break;
default:
break;
}
}
{
// rebuildList contains the user-selected index(in USER_DEFINED mode).
{
{
// Get all existing indexes for all && degraded mode.
}
else if (!rebuildList.isEmpty())
{
// Get indexes for user defined index.
{
{
}
}
}
}
}
private void rebuildAttributeIndexes(WriteableTransaction txn, AttributeIndex attrIndex, AttributeType attrType,
boolean onlyDegraded) throws StorageRuntimeException
{
{
}
}
private void fillIndexMap(WriteableTransaction txn, AttributeType attrType, MatchingRuleIndex index,
{
{
}
}
private void clearIndexes(WriteableTransaction txn, boolean onlyDegraded) throws StorageRuntimeException
{
// Clears all the entry's container databases which are containing the indexes
if (!onlyDegraded)
{
// dn2uri does not have a trusted status.
}
if (!onlyDegraded
{
}
{
{
}
}
{
{
}
}
}
private void setRebuildListIndexesTrusted(WriteableTransaction txn, boolean trusted) throws StorageRuntimeException
{
try
{
{
}
{
}
}
catch (StorageRuntimeException ex)
{
throw new StorageRuntimeException(NOTE_JEB_IMPORT_LDIF_TRUSTED_FAILED.get(ex.getMessage()).toString());
}
}
private void setTrusted(WriteableTransaction txn, final Collection<MatchingRuleIndex> indexes, boolean trusted)
{
{
{
}
}
}
/** @see Importer#importPhaseOne(WriteableTransaction) */
{
for (int i = 0; i < threadCount; i++)
{
}
// Try to clear as much memory as possible.
}
{
try
{
}
finally
{
}
}
{
return timer;
}
{
switch (rebuildConfig.getRebuildMode())
{
case ALL:
return getTotalIndexCount(cfg);
case DEGRADED:
// FIXME: since the environment is not started 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;
}
{
{
}
{
}
}
{
{
}
}
{
{
{
}
}
}
/**
* Return the number of entries processed by the rebuild manager.
*
* @return The number of entries processed.
*/
public long getEntriesProcess()
{
return this.entriesProcessed.get();
}
/**
* Return the total number of entries to process by the rebuild manager.
*
* @return The total number for entries to process.
*/
public 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 database.
*/
public RebuildFirstPhaseProgressTask() throws StorageRuntimeException
{
}
/**
* The action to be performed by this timer task.
*/
public void run()
{
if (deltaTime == 0)
{
return;
}
float completed = 0;
{
}
}
}
/**
* 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;
private long latestCount;
/**
* Create a new import progress task.
*
* @param latestCount
* The latest count of entries processed in phase one.
*/
public SecondPhaseProgressTask(long latestCount)
{
this.latestCount = latestCount;
}
/** 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.
{
}
}
}
/**
* A class to hold information about the entry determined by the LDIF reader.
* Mainly the suffix the entry belongs under and the ID assigned to it by the
* reader.
*/
public class EntryInformation
{
/**
* Return the suffix associated with the entry.
*
* @return Entry's suffix instance;
*/
{
return suffix;
}
/**
* Set the suffix instance associated with the entry.
*
* @param suffix
* The suffix associated with the entry.
*/
{
}
/**
* Set the entry's ID.
*
* @param entryID
* The entry ID to set the entry ID to.
*/
{
}
/**
* Return the entry ID associated with the entry.
*
* @return The entry ID associated with the entry.
*/
private EntryID getEntryID()
{
return entryID;
}
}
/**
* This class is used as an index key for hash maps that need to process multiple suffix index
* cn.equality, sn.equality,...).
*/
public 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()
+ "(attributeType=" + attributeType
+ ", indexID=" + indexID
+ ", entryLimit=" + entryLimit
+ ")";
}
}
/**
* The temporary environment will be shared when multiple suffixes are being
* processed. This interface is used by those suffix instance to do parental
* checking of the DN cache.
*/
public static interface DNCache
{
/**
* Returns {@code true} if the specified DN is contained in the DN cache, or
* {@code false} otherwise.
*
* @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 database.
*/
}
/** 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
}
}
/**
* Temporary environment used to check DN's when DN validation is performed
* during phase one processing. It is deleted after phase one processing.
*/
{
/**
* Create a temporary DB environment and database to be used as a cache of
* DNs when DN validation is performed in phase one processing.
*
* @param envPath
* The file path to create the environment under.
* @throws StorageRuntimeException
* If an error occurs either creating the environment or the DN database.
*/
{
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++)
{
}
}
/**
* Shutdown the temporary environment.
*
* @throws StorageRuntimeException
* If error occurs.
*/
public void shutdown() throws StorageRuntimeException
{
try
{
}
finally
{
}
}
/**
* 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 database.
*/
{
// 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;
}
}));
}
});
}
catch (StorageRuntimeException e)
{
throw e;
}
catch (Exception e)
{
throw new StorageRuntimeException(e);
}
return updateResult.get();
}
/** 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;
}
/**
* Check if the specified DN is contained in the temporary DN cache.
*
* @param dn
* A DN check for.
* @return {@code true} if the specified DN is in the temporary DN cache, or
* {@code false} if it is not.
*/
{
try
{
{
{
if (existingDns != null)
{
}
return false;
}
});
}
catch (StorageRuntimeException e)
{
throw e;
}
catch (Exception e)
{
throw new StorageRuntimeException(e);
}
}
}
}