package org.mapdb;
import java.io.DataInput;
import java.io.InputStream;
import java.io.OutputStream;
import java.util.Arrays;
import java.util.Collections;
import java.util.HashSet;
import java.util.Set;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.atomic.AtomicLong;
import java.util.concurrent.locks.Lock;
import java.util.logging.Level;
/**
* append only store
*/
public class StoreAppend extends Store {
/** 2 byte store version*/
protected static final int STORE_VERSION = 100;
/** 4 byte file header */
protected static final int HEADER = (0xAB3D<<16) | STORE_VERSION;
protected static final int I_UPDATE = 1;
protected static final int I_INSERT = 3;
protected static final int I_DELETE = 2;
protected static final int I_PREALLOC = 4;
protected static final int I_SKIP_SINGLE_BYTE = 6;
protected static final int I_SKIP_MULTI_BYTE = 7;
protected static final int I_TX_VALID = 8;
protected static final int I_TX_ROLLBACK = 9;
protected static final long headerSize = 16;
protected static final StoreAppend[] STORE_APPENDS_ZERO_ARRAY = new StoreAppend[0];
protected Volume vol;
/**
* In memory table which maps recids into their offsets. Positive values are offsets.
* Zero value indicates on-used records
* Negative values are:
*
* -1 - records was deleted, return null
* -2 - record has zero size
* -3 - null record, return null
*
*
*
*/
//TODO this is in-memory, move to temporary file or something
protected Volume indexTable;
//guarded by StructuralLock
protected long eof = 0;
protected final AtomicLong highestRecid = new AtomicLong(0);
protected final boolean tx;
protected final LongLongMap[] modified;
protected final ScheduledExecutorService compactionExecutor;
protected final Set snapshots;
protected final boolean isSnapshot;
protected final long startSize;
protected final long sizeIncrement;
protected final int sliceShift;
protected StoreAppend(String fileName,
Volume.VolumeFactory volumeFactory,
Cache cache,
int lockScale,
int lockingStrategy,
boolean checksum,
boolean compress,
byte[] password,
boolean readonly,
boolean snapshotEnable,
boolean fileLockDisable,
DataIO.HeartbeatFileLock fileLockHeartbeat,
boolean txDisabled,
ScheduledExecutorService compactionExecutor,
long startSize,
long sizeIncrement
) {
super(fileName, volumeFactory, cache, lockScale,lockingStrategy, checksum, compress, password, readonly,
snapshotEnable,fileLockDisable, fileLockHeartbeat);
this.tx = !txDisabled;
if(tx){
modified = new LongLongMap[this.lockScale];
for(int i=0;i());
this.isSnapshot = false;
this.sizeIncrement = Math.max(1L<=volumeSize)
break;
final long instPos = pos;
final int inst = vol.getUnsignedByte(pos++);
if (inst == I_INSERT || inst == I_UPDATE) {
long recid = vol.getPackedLong(pos);
pos += recid>>>60;
recid = longParityGet(recid & DataIO.PACK_LONG_RESULT_MASK);
highestRecid2 = Math.max(highestRecid2, recid);
commitData.put(recid, instPos);
//skip rest of the record
long size = vol.getPackedLong(pos);
long dataLen = longParityGet(size & DataIO.PACK_LONG_RESULT_MASK) - 1;
dataLen = Math.max(0,dataLen);
pos = pos + (size>>>60) + dataLen;
} else if (inst == I_DELETE) {
long recid = vol.getPackedLong(pos);
pos += recid>>>60;
recid = longParityGet(recid & DataIO.PACK_LONG_RESULT_MASK);
highestRecid2 = Math.max(highestRecid2, recid);
commitData.put(recid, -1);
} else if (inst == I_DELETE) {
long recid = vol.getPackedLong(pos);
pos += recid>>>60;
recid = longParityGet(recid & DataIO.PACK_LONG_RESULT_MASK);
highestRecid2 = Math.max(highestRecid2, recid);
commitData.put(recid,-2);
} else if (inst == I_SKIP_SINGLE_BYTE) {
//do nothing, just skip single byte
} else if (inst == I_SKIP_MULTI_BYTE) {
//read size and skip it
//skip rest of the record
long size = vol.getPackedLong(pos);
pos += (size>>>60) + longParityGet(size & DataIO.PACK_LONG_RESULT_MASK);
} else if (inst == I_TX_VALID) {
if (tx){
lastValidCommitOffset = pos;
//apply changes from commitData to indexTable
for(int i=0;i A get2(long recid, Serializer serializer) {
if(CC.ASSERT)
assertReadLocked(recid);
long offset = tx?
modified[lockPos(recid)].get(recid):
0;
if(offset==0) {
try {
offset = indexTable.getLong(recid * 8);
} catch (ArrayIndexOutOfBoundsException e) {
//TODO this code should be aware if indexTable internals?
throw new DBException.EngineGetVoid();
}
}
if(offset==-3||offset==-1) //null, preallocated or deleted
return null;
if(offset == 0){ //non existent
throw new DBException.EngineGetVoid();
}
if(offset == -2){
//zero size record
return deserialize(serializer,0,new DataIO.DataInputByteArray(new byte[0]));
}
final long packedRecidSize = DataIO.packLongSize(longParitySet(recid));
if(CC.ASSERT){
int instruction = vol.getUnsignedByte(offset);
if(instruction!= I_UPDATE && instruction!= I_INSERT)
throw new DBException.DataCorruption("wrong instruction "+instruction);
long recid2 = vol.getPackedLong(offset+1);
if(packedRecidSize!=recid2>>>60)
throw new DBException.DataCorruption("inconsistent recid len");
recid2 = longParityGet(recid2&DataIO.PACK_LONG_RESULT_MASK);
if(recid!=recid2)
throw new DBException.DataCorruption("recid does not match");
}
offset += 1 + //instruction size
packedRecidSize; // recid size
//read size
long size = vol.getPackedLong(offset);
offset+=size>>>60;
size = longParityGet(size & DataIO.PACK_LONG_RESULT_MASK);
size -= 1; //normalize size
if(CC.ASSERT && size<=0)
throw new DBException.DataCorruption("wrong size");
DataInput input = vol.getDataInputOverlap(offset, (int) size);
return deserialize(serializer, (int) size, input);
}
@Override
protected void update2(long recid, DataIO.DataOutputByteArray out) {
insertOrUpdate(recid, out, false);
}
private void insertOrUpdate(long recid, DataIO.DataOutputByteArray out, boolean isInsert) {
if(CC.ASSERT)
assertWriteLocked(lockPos(recid));
//TODO assert indexTable state, record should already exist/not exist
final int realSize = out==null ? 0: out.pos;
final int shiftedSize = out==null ?0 : realSize+1; //one additional state to indicate null
final int headSize = 1 + //instruction
DataIO.packLongSize(longParitySet(recid)) + //recid
DataIO.packLongSize(longParitySet(shiftedSize)); //length
long offset = alloc(headSize, headSize+realSize);
final long origOffset = offset;
//ensure available worst case scenario
vol.ensureAvailable(offset+headSize+realSize);
//instruction
vol.putUnsignedByte(offset, isInsert ? I_INSERT : I_UPDATE);
offset++;
//recid
offset+=vol.putPackedLong(offset,longParitySet(recid));
//size
offset+=vol.putPackedLong(offset,longParitySet(shiftedSize));
if(realSize!=0)
vol.putDataOverlap(offset, out.buf,0,out.pos);
// -3 is null record
// -2 is zero size record
indexTablePut(recid, out==null? -3 : (realSize==0) ? -2:origOffset);
}
@Override
protected void delete2(long recid, Serializer serializer) {
if(CC.ASSERT)
assertWriteLocked(lockPos(recid));
final int headSize = 1 + DataIO.packLongSize(longParitySet(recid));
long offset = alloc(headSize,headSize);
vol.ensureAvailable(offset + headSize);
vol.putUnsignedByte(offset, I_DELETE); //delete instruction
offset++;
vol.putPackedLong(offset,longParitySet(recid));
indexTablePut(recid, -1); // -1 is deleted record
}
@Override
public long getCurrSize() {
return 0;
}
@Override
public long getFreeSize() {
return 0;
}
@Override
public boolean fileLoad() {
return vol.fileLoad();
}
@Override
public long preallocate() {
long recid = highestRecid.incrementAndGet();
Lock lock = locks[lockPos(recid)].writeLock();
lock.lock();
try{
final int headSize = 1 + DataIO.packLongSize(longParitySet(recid));
long offset = alloc(headSize,headSize);
vol.ensureAvailable(offset + headSize);
vol.putUnsignedByte(offset, I_PREALLOC);
offset++;
vol.putPackedLong(offset, longParitySet(recid));
indexTablePut(recid,-3);
}finally {
lock.unlock();
}
return recid;
}
protected void indexTablePut(long recid, long offset) {
if(tx){
modified[lockPos(recid)].put(recid,offset);
}else {
indexTable.ensureAvailable(recid*8+8);
indexTable.putLong(recid * 8, offset);
}
}
@Override
public long put(A value, Serializer serializer) {
DataIO.DataOutputByteArray out = serialize(value,serializer);
long recid = highestRecid.incrementAndGet();
int lockPos = lockPos(recid);
Cache cache = caches==null ? null : caches[lockPos] ;
Lock lock = locks[lockPos].writeLock();
lock.lock();
try{
if(cache!=null) {
cache.put(recid, value);
}
insertOrUpdate(recid,out,true);
}finally {
lock.unlock();
}
return recid;
}
@Override
public void close() {
if(closed)
return;
commitLock.lock();
try {
if(closed)
return;
if(isSnapshot){
snapshots.remove(this);
return;
}
vol.sync();
vol.close();
indexTable.close();
if(caches!=null){
for(Cache c:caches){
c.close();
}
Arrays.fill(caches,null);
}
if(fileLockHeartbeat !=null) {
fileLockHeartbeat.unlock();
fileLockHeartbeat = null;
}
closed = true;
}finally{
commitLock.unlock();
}
}
@Override
public void commit() {
if(isSnapshot)
return;
if(!tx){
vol.sync();
return;
}
commitLock.lock();
try{
StoreAppend[] snaps = snapshots==null ?
STORE_APPENDS_ZERO_ARRAY :
snapshots.toArray(STORE_APPENDS_ZERO_ARRAY);
for(int i=0;i
Ads help maintain this website.