/*
|
* 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
|
* or http://forgerock.org/license/CDDLv1.0.html.
|
* 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 2010 Sun Microsystems, Inc.
|
* Portions Copyright 2012-2015 ForgeRock AS.
|
*/
|
package org.opends.server.backends.pluggable;
|
|
import static org.opends.messages.BackendMessages.*;
|
import static org.opends.server.util.StaticUtils.*;
|
|
import java.io.IOException;
|
import java.nio.ByteBuffer;
|
import java.nio.channels.FileChannel;
|
|
import org.forgerock.i18n.slf4j.LocalizedLogger;
|
import org.forgerock.opendj.ldap.ByteString;
|
import org.forgerock.opendj.ldap.ByteStringBuilder;
|
import org.opends.server.backends.pluggable.OnDiskMergeBufferImporter.IndexManager;
|
|
/**
|
* The buffer class is used to process a buffer from the temporary index files
|
* during phase 2 processing.
|
*/
|
final class IndexInputBuffer implements Comparable<IndexInputBuffer>
|
{
|
|
/** Possible states while reading a record. */
|
private static enum RecordState
|
{
|
START, NEED_INSERT_ID_SET, NEED_DELETE_ID_SET
|
}
|
|
private static final LocalizedLogger logger = LocalizedLogger.getLoggerForThisClass();
|
static final long UNDEFINED_SIZE = -1;
|
|
private final IndexManager indexMgr;
|
private final FileChannel channel;
|
private final long begin;
|
private final long end;
|
private final int bufferID;
|
|
private long offset;
|
private final ByteBuffer cache;
|
|
/** Next fields are the fetched record data. */
|
private ImportRecord record;
|
private final ByteStringBuilder keyBuffer = new ByteStringBuilder(128);
|
private RecordState recordState = RecordState.START;
|
|
/**
|
* Creates a new index input buffer.
|
*
|
* @param indexMgr
|
* The index manager.
|
* @param channel
|
* The index file channel.
|
* @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 bufferID
|
* The buffer ID.
|
* @param cacheSize
|
* The cache size.
|
* @throws IOException
|
* If an IO error occurred when priming the cache.
|
*/
|
IndexInputBuffer(IndexManager indexMgr, FileChannel channel, long begin, long end, int bufferID, int cacheSize)
|
throws IOException
|
{
|
this.indexMgr = indexMgr;
|
this.channel = channel;
|
this.begin = begin;
|
this.end = end;
|
this.offset = 0;
|
this.bufferID = bufferID;
|
this.cache = ByteBuffer.allocate(Math.max(cacheSize - 384, 256));
|
|
loadCache();
|
cache.flip();
|
}
|
|
private void loadCache() throws IOException
|
{
|
channel.position(begin + offset);
|
long leftToRead = end - (begin + offset);
|
long bytesToRead;
|
if (leftToRead < cache.remaining())
|
{
|
cache.limit((int) (cache.position() + leftToRead));
|
bytesToRead = leftToRead;
|
}
|
else
|
{
|
bytesToRead = Math.min((end - offset), cache.remaining());
|
}
|
int bytesRead = 0;
|
while (bytesRead < bytesToRead)
|
{
|
bytesRead += channel.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.
|
*/
|
boolean hasMoreData() throws IOException
|
{
|
boolean hasMore = begin + offset < end;
|
return cache.remaining() != 0 || hasMore;
|
}
|
|
ImportRecord currentRecord()
|
{
|
if (record == null)
|
{
|
// ensure record fetched
|
try
|
{
|
fetchNextRecord();
|
}
|
catch (IOException ex)
|
{
|
logger.error(ERR_IMPORT_BUFFER_IO_ERROR, indexMgr.getBufferFileName());
|
throw new RuntimeException(ex);
|
}
|
}
|
return record;
|
}
|
|
/**
|
* Reads the next record from the buffer, skipping any remaining data in the
|
* current record.
|
*
|
* @throws IOException
|
* If an IO error occurred.
|
*/
|
void fetchNextRecord() 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;
|
}
|
|
int indexID = getInt();
|
ByteString key = toKey();
|
record = ImportRecord.from(key, indexID);
|
|
recordState = RecordState.NEED_INSERT_ID_SET;
|
}
|
|
private ByteString toKey() throws IOException
|
{
|
ensureData(20);
|
int keyLen = getInt();
|
ensureData(keyLen);
|
keyBuffer.clear().append(cache, keyLen);
|
return keyBuffer.toByteString();
|
}
|
|
private int getInt() throws IOException
|
{
|
ensureData(INT_SIZE);
|
return cache.getInt();
|
}
|
|
private long getLong() throws IOException
|
{
|
ensureData(LONG_SIZE);
|
return cache.getLong();
|
}
|
|
/**
|
* 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.
|
*/
|
void mergeIDSet(ImportIDSet idSet) throws IOException
|
{
|
if (recordState == RecordState.START)
|
{
|
throw new IllegalStateException();
|
}
|
|
ensureData(20);
|
int keyCount = getInt();
|
for (int k = 0; k < keyCount; k++)
|
{
|
long entryID = getLong();
|
|
// idSet will be null if skipping.
|
if (idSet != null)
|
{
|
if (entryID == UNDEFINED_SIZE)
|
{
|
idSet.setUndefined();
|
}
|
else
|
{
|
idSet.addEntryID(entryID);
|
}
|
}
|
}
|
|
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
|
{
|
if (cache.remaining() == 0)
|
{
|
cache.clear();
|
loadCache();
|
cache.flip();
|
return true;
|
}
|
else if (cache.remaining() < len)
|
{
|
cache.compact();
|
loadCache();
|
cache.flip();
|
return true;
|
}
|
return false;
|
}
|
|
/** {@inheritDoc} */
|
@Override
|
public int compareTo(IndexInputBuffer o)
|
{
|
// used in remove.
|
if (this == o)
|
{
|
return 0;
|
}
|
|
int cmp = currentRecord().compareTo(o.currentRecord());
|
if (cmp == 0)
|
{
|
return bufferID - o.bufferID;
|
}
|
return cmp;
|
}
|
|
/** {@inheritDoc} */
|
@Override
|
public String toString()
|
{
|
return getClass().getSimpleName() + "("
|
+ ", indexMgr=" + indexMgr + "bufferID=" + bufferID + ", record=" + record + ", recordState=" + recordState
|
+ ")";
|
}
|
}
|
