package org.mapdb;
import java.io.*;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.BitSet;
import java.util.List;
import java.util.concurrent.CopyOnWriteArrayList;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.Future;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.atomic.AtomicLong;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReadWriteLock;
import java.util.logging.Level;
import static org.mapdb.DataIO.*;
public class StoreDirect extends Store {
/** 2 byte store version*/
protected static final int STORE_VERSION = 100;
/** 4 byte file header */
protected static final int HEADER = (0xA9DB<<16) | STORE_VERSION;
protected static final long PAGE_SIZE = 1<< CC.VOLUME_PAGE_SHIFT;
protected static final long PAGE_MASK = PAGE_SIZE-1;
protected static final long PAGE_MASK_INVERSE = 0xFFFFFFFFFFFFFFFFL< snapshots;
protected final long indexValSize;
protected final long startSize;
protected final long sizeIncrement;
protected final boolean recidReuseDisable;
protected final int sliceShift;
protected final AtomicLong freeSize = new AtomicLong(-1);
public StoreDirect(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,
ScheduledExecutorService executor,
long startSize,
long sizeIncrement,
boolean recidReuseDisable
) {
super(fileName, volumeFactory, cache, lockScale, lockingStrategy, checksum, compress, password, readonly,
snapshotEnable, fileLockDisable, fileLockHeartbeat);
this.executor = executor;
this.snapshots = snapshotEnable?
new CopyOnWriteArrayList():
null;
this.indexValSize = checksum ? 10 : 8;
this.sizeIncrement = Math.max(1L< A get2(long recid, Serializer serializer) {
if (CC.ASSERT)
assertReadLocked(recid);
long[] offsets = offsetsGet(lockPos(recid),indexValGet(recid));
return getFromOffset(serializer, offsets);
}
protected A getFromOffset(Serializer serializer, long[] offsets) {
if (CC.LOG_STORE && LOG.isLoggable(Level.FINEST)) {
LOG.log(Level.FINEST, "serializer={0}, offsets={1}",new Object[]{serializer, Arrays.toString(offsets)});
}
if (offsets == null) {
return null; //zero size
}else if (offsets.length==0){
return deserialize(serializer,0,new DataInputByteArray(new byte[0]));
}else if (offsets.length == 1) {
//not linked
int size = (int) (offsets[0] >>> 48);
long offset = offsets[0] & MOFFSET;
DataInput in = vol.getDataInput(offset, size);
return deserialize(serializer, size, in);
} else {
//calculate total size
int totalSize = offsetsTotalSize(offsets);
byte[] b = getLoadLinkedRecord(offsets, totalSize);
DataInput in = new DataInputByteArray(b);
return deserialize(serializer, totalSize, in);
}
}
private byte[] getLoadLinkedRecord(long[] offsets, int totalSize) {
if (CC.LOG_STORE && LOG.isLoggable(Level.FINEST)) {
LOG.log(Level.FINEST, "totalSize={0}, offsets={1}", new Object[]{totalSize, Arrays.toString(offsets)});
}
//load data
byte[] b = new byte[totalSize];
int bpos = 0;
for (int i = 0; i < offsets.length; i++) {
int plus = (i == offsets.length - 1)?0:8;
long size = (offsets[i] >>> 48) - plus;
if(CC.ASSERT && (size&0xFFFF)!=size)
throw new DBException.DataCorruption("size mismatch");
long offset = offsets[i] & MOFFSET;
//System.out.println("GET "+(offset + plus)+ " - "+size+" - "+bpos);
vol.getData(offset + plus, b, bpos, (int) size);
bpos += size;
}
if (CC.ASSERT && bpos != totalSize)
throw new DBException.DataCorruption("size does not match");
return b;
}
protected int offsetsTotalSize(long[] offsets) {
if(offsets==null || offsets.length==0)
return 0;
int totalSize = 8;
for (long l : offsets) {
totalSize += (l >>> 48) - 8;
}
return totalSize;
}
@Override
protected void update2(long recid, DataOutputByteArray out) {
int pos = lockPos(recid);
if(CC.ASSERT)
assertWriteLocked(pos);
long oldIndexVal = indexValGet(recid);
boolean releaseOld = true;
if(snapshotEnable){
for(Snapshot snap:snapshots){
snap.oldRecids[pos].putIfAbsent(recid,oldIndexVal);
releaseOld = false;
}
}
long[] oldOffsets = offsetsGet(pos,oldIndexVal);
int oldSize = offsetsTotalSize(oldOffsets);
int newSize = out==null?0:out.pos;
long[] newOffsets;
if (CC.LOG_STORE && LOG.isLoggable(Level.FINEST)) {
LOG.log(Level.FINEST, "recid={0}, oldIndexVal={1}, oldSize={2}, newSize={3}, oldOffsets={4}",
new Object[]{recid, oldIndexVal, oldSize, newSize, Arrays.toString(oldOffsets)});
}
//if new version fits into old one, reuse space
if(releaseOld && oldSize==newSize){
//TODO more precise check of linked records
//TODO check rounUp 16 for non-linked records
newOffsets = oldOffsets;
}else {
structuralLock.lock();
try {
if(releaseOld && oldOffsets!=null)
freeDataPut(oldOffsets);
newOffsets = newSize==0?null:freeDataTake(out.pos);
} finally {
structuralLock.unlock();
}
}
if(CC.ASSERT)
offsetsVerify(newOffsets);
putData(recid, newOffsets, out==null?null:out.buf, out==null?0:out.pos);
}
protected void offsetsVerify(long[] ret) {
//TODO check non tail records are mod 16
//TODO check linkage
if(ret==null)
return;
for(int i=0;i>>48);
if(size<=0)
throw new DBException.DataCorruption("size too small");
}
}
/** return positions of (possibly) linked record */
protected long[] offsetsGet(int segment, long indexVal) {;
if(indexVal>>>48==0){
return ((indexVal&MLINKED)!=0) ? null : EMPTY_LONGS;
}
long[] ret = new long[]{indexVal};
while((ret[ret.length-1]&MLINKED)!=0){
ret = Arrays.copyOf(ret,ret.length+1);
ret[ret.length-1] = parity3Get(vol.getLong(ret[ret.length-2]&MOFFSET));
}
if(CC.ASSERT){
offsetsVerify(ret);
}
if (CC.LOG_STORE && LOG.isLoggable(Level.FINEST)) {
LOG.log(Level.FINEST, "indexVal={0}, ret={1}",
new Object[]{Long.toHexString(indexVal), Arrays.toString(ret)});
}
return ret;
}
protected void indexValPut(long recid, int size, long offset, boolean linked, boolean unused) {
if(CC.ASSERT)
assertWriteLocked(lockPos(recid));
long indexOffset = recidToOffset(recid);
long newval = composeIndexVal(size, offset, linked, unused, true);
if (CC.LOG_STORE && LOG.isLoggable(Level.FINEST)) {
LOG.log(Level.FINEST, "recid={0}, indexOffset={1}, newval={2}",
new Object[]{recid, indexOffset, Long.toHexString(newval)});
}
vol.putLong(indexOffset, newval);
if(checksum){
vol.putUnsignedShort(indexOffset + 8, DataIO.longHash(newval) & 0xFFFF);
}
}
@Override
protected void delete2(long recid, Serializer serializer) {
if(CC.ASSERT)
assertWriteLocked(lockPos(recid));
final int pos = lockPos(recid);
long oldIndexVal = indexValGet(recid);
long[] offsets = offsetsGet(pos,oldIndexVal);
boolean releaseOld = true;
if(snapshotEnable){
for(Snapshot snap:snapshots){
snap.oldRecids[pos].putIfAbsent(recid,oldIndexVal);
releaseOld = false;
}
}
if (CC.LOG_STORE && LOG.isLoggable(Level.FINEST)) {
LOG.log(Level.FINEST, "recid={0}, oldIndexVal={1}, releaseOld={2}, offsets={3}",
new Object[]{recid, Long.toHexString(oldIndexVal), releaseOld, Arrays.toString(offsets)});
}
if(offsets!=null && releaseOld) {
structuralLock.lock();
try {
freeDataPut(offsets);
} finally {
structuralLock.unlock();
}
}
indexValPut(recid, 0, 0, true, true);
if(!recidReuseDisable){
structuralLock.lock();
try {
longStackPut(FREE_RECID_STACK, recid, false);
}finally {
structuralLock.unlock();
}
}
}
@Override
public long getCurrSize() {
return vol.length() - lastAllocatedData % CHUNKSIZE;
}
@Override
public long getFreeSize() {
long ret = freeSize.get();
if(ret!=-1)
return ret;
structuralLock.lock();
try{
//try one more time under lock
ret = freeSize.get();
if(ret!=-1)
return ret;
//traverse list of recids,
ret=
8* longStackCount(FREE_RECID_STACK);
for(long stackNum = 1;stackNum<=SLOTS_COUNT;stackNum++){
long indexOffset = FREE_RECID_STACK+stackNum*8;
long size = stackNum*16;
ret += size * longStackCount(indexOffset);
}
freeSize.set(ret);
return ret;
}finally {
structuralLock.unlock();
}
}
@Override
public boolean fileLoad() {
return vol.fileLoad();
}
protected void freeSizeIncrement(int increment){
for(;;) {
long val = freeSize.get();
if (val == -1 || freeSize.compareAndSet(val, val + increment))
return;
}
}
@Override
public long preallocate() {
long recid;
structuralLock.lock();
try {
//TODO possible race condition here? Can this modify existing data?
recid = freeRecidTake();
}finally {
structuralLock.unlock();
}
Lock lock = locks[lockPos(recid)].writeLock();
lock.lock();
try {
indexValPut(recid, 0, 0L, true, true);
}finally {
lock.unlock();
}
if (CC.LOG_STORE && LOG.isLoggable(Level.FINEST)) {
LOG.log(Level.FINEST, "recid={0}",recid);
}
return recid;
}
@Override
public long put(A value, Serializer serializer) {
long recid;
long[] offsets;
DataOutputByteArray out = serialize(value, serializer);
boolean notalloc = out==null || out.pos==0;
commitLock.lock();
try {
structuralLock.lock();
try {
recid = freeRecidTake();
} finally {
structuralLock.unlock();
}
int pos = lockPos(recid);
Lock lock = locks[pos].writeLock();
lock.lock();
//TODO possible deadlock, should not lock segment under different segment lock
//TODO investigate if this lock is necessary, recid has not been yet published, perhaps cache does not have to be updated
try {
if(CC.ASSERT && recidReuseDisable && vol.getLong(recidToOffset(recid))!=0){
throw new AssertionError("Recid not empty: "+recid);
}
if (caches != null) {
caches[pos].put(recid, value);
}
if (snapshotEnable) {
for (Snapshot snap : snapshots) {
snap.oldRecids[pos].putIfAbsent(recid, 0);
}
}
structuralLock.lock();
try {
offsets = notalloc ? null : freeDataTake(out.pos);
} finally {
structuralLock.unlock();
}
if (CC.ASSERT && offsets != null && (offsets[0] & MOFFSET) < PAGE_SIZE)
throw new DBException.DataCorruption();
putData(recid, offsets, out == null ? null : out.buf, out == null ? 0 : out.pos);
} finally {
lock.unlock();
}
}finally {
commitLock.unlock();
}
if (CC.LOG_STORE && LOG.isLoggable(Level.FINEST)) {
LOG.log(Level.FINEST, "recid={0}, serSize={1}, serializer={2}",
new Object[]{recid, notalloc?0:out.pos, serializer});
}
return recid;
}
protected void putData(long recid, long[] offsets, byte[] src, int srcLen) {
if(CC.ASSERT)
assertWriteLocked(lockPos(recid));
if(CC.ASSERT && offsetsTotalSize(offsets)!=(src==null?0:srcLen))
throw new DBException.DataCorruption("size mismatch");
if (CC.LOG_STORE && LOG.isLoggable(Level.FINEST)) {
LOG.log(Level.FINEST, "recid={0}, srcLen={1}, offsets={2}",
new Object[]{recid, srcLen, Arrays.toString(offsets)});
}
if(offsets!=null) {
int outPos = 0;
for (int i = 0; i < offsets.length; i++) {
final boolean last = (i == offsets.length - 1);
if (CC.ASSERT && ((offsets[i] & MLINKED) == 0) != last)
throw new DBException.DataCorruption("linked bit set wrong way");
long offset = (offsets[i] & MOFFSET);
if(CC.ASSERT && offset%16!=0)
throw new DBException.DataCorruption("not aligned to 16");
int plus = (last?0:8);
int size = (int) ((offsets[i]>>>48) - plus);
if(CC.ASSERT && ((size&0xFFFF)!=size || size==0))
throw new DBException.DataCorruption("size mismatch");
int segment = lockPos(recid);
//write offset to next page
if (!last) {
putDataSingleWithLink(segment, offset,parity3Set(offsets[i + 1]), src,outPos,size);
}else{
putDataSingleWithoutLink(segment, offset, src, outPos, size);
}
outPos += size;
}
if(CC.ASSERT && outPos!=srcLen)
throw new DBException.DataCorruption("size mismatch");
}
//update index val
boolean firstLinked =
(offsets!=null && offsets.length>1) || //too large record
(src==null); //null records
boolean empty = offsets==null || offsets.length==0;
int firstSize = (int) (empty ? 0L : offsets[0]>>>48);
long firstOffset = empty? 0L : offsets[0]&MOFFSET;
indexValPut(recid, firstSize, firstOffset, firstLinked, false);
}
protected void putDataSingleWithoutLink(int segment, long offset, byte[] buf, int bufPos, int size) {
vol.putData(offset, buf, bufPos, size);
}
protected void putDataSingleWithLink(int segment, long offset, long link, byte[] buf, int bufPos, int size) {
vol.putLong(offset, link);
vol.putData(offset + 8, buf, bufPos, size);
}
protected void freeDataPut(long[] linkedOffsets) {
if(CC.ASSERT && !structuralLock.isHeldByCurrentThread())
throw new AssertionError();
for(long v:linkedOffsets){
int size = round16Up((int) (v >>> 48));
v &= MOFFSET;
freeDataPut(v,size);
}
}
protected void freeDataPut(long offset, int size) {
if(CC.ASSERT && !structuralLock.isHeldByCurrentThread())
throw new AssertionError();
if(CC.ASSERT && size%16!=0 )
throw new DBException.DataCorruption("unalligned size");
if(CC.ASSERT && (offset%16!=0 || offset>> 4, //offset is multiple of 16, save some space
false);
}
protected long[] freeDataTake(int size) {
if(CC.ASSERT && !structuralLock.isHeldByCurrentThread())
throw new AssertionError();
if(CC.ASSERT && size<=0)
throw new DBException.DataCorruption("size too small");
//compose of multiple single records
long[] ret = EMPTY_LONGS;
while(size>MAX_REC_SIZE){
ret = Arrays.copyOf(ret,ret.length+1);
ret[ret.length-1] = (((long)MAX_REC_SIZE)<<48) | freeDataTakeSingle(round16Up(MAX_REC_SIZE)) | MLINKED;
size = size-MAX_REC_SIZE+8;
}
//allocate last section
ret = Arrays.copyOf(ret,ret.length+1);
ret[ret.length-1] = (((long)size)<<48) | freeDataTakeSingle(round16Up(size)) ;
if (CC.LOG_STORE && LOG.isLoggable(Level.FINEST)) {
LOG.log(Level.FINEST, "size={0}, ret={1}",
new Object[]{size, Arrays.toString(ret)});
}
return ret;
}
protected long freeDataTakeSingle(int size) {
if(CC.ASSERT && !structuralLock.isHeldByCurrentThread())
throw new AssertionError();
if(CC.ASSERT && size%16!=0)
throw new DBException.DataCorruption("unalligned size");
if(CC.ASSERT && size>round16Up(MAX_REC_SIZE))
throw new DBException.DataCorruption("size too big");
long masterPointerOffset = size/2 + FREE_RECID_STACK; // really is size*8/16
long ret = longStackTake(masterPointerOffset,false) <<4; //offset is multiple of 16, save some space
if(ret!=0) {
if(CC.ASSERT && retPAGE_SIZE || masterLinkOffset % 8!=0)) //TODO perhaps remove the last check
throw new DBException.DataCorruption("wrong master link");
long masterLinkVal = parity4Get(headVol.getLong(masterLinkOffset));
long pageOffset = masterLinkVal&MOFFSET;
if(masterLinkVal==0L){
longStackNewPage(masterLinkOffset, 0L, value);
return;
}
long currSize = masterLinkVal>>>48;
long prevLinkVal = parity4Get(vol.getLong(pageOffset));
long pageSize = prevLinkVal>>>48;
//is there enough space in current page?
if(currSize+8>=pageSize){ // +8 is just to make sure and is worse case scenario, perhaps make better check based on actual packed size
//no there is not enough space
//first zero out rest of the page
vol.clear(pageOffset+currSize, pageOffset+pageSize);
//allocate new page
longStackNewPage(masterLinkOffset,pageOffset,value);
return;
}
//there is enough space, so just write new value
currSize += vol.putLongPackBidi(pageOffset+currSize, longParitySet(value));
//and update master pointer
headVol.putLong(masterLinkOffset, parity4Set(currSize<<48 | pageOffset));
}
protected void longStackNewPage(long masterLinkOffset, long prevPageOffset, long value) {
if(CC.ASSERT && !structuralLock.isHeldByCurrentThread())
throw new AssertionError();
long newPageOffset = freeDataTakeSingle((int) CHUNKSIZE);
//write size of current chunk with link to prev page
vol.putLong(newPageOffset, parity4Set((CHUNKSIZE<<48) | prevPageOffset));
//put value
long currSize = 8 + vol.putLongPackBidi(newPageOffset+8, longParitySet(value));
//update master pointer
headVol.putLong(masterLinkOffset, parity4Set((currSize<<48)|newPageOffset));
}
protected long longStackTake(long masterLinkOffset, boolean recursive){
if(CC.ASSERT && !structuralLock.isHeldByCurrentThread())
throw new AssertionError();
if(CC.ASSERT && (masterLinkOffsetFREE_RECID_STACK+round16Up(MAX_REC_SIZE)/2 ||
masterLinkOffset % 8!=0))
throw new DBException.DataCorruption("wrong master link");
long masterLinkVal = parity4Get(headVol.getLong(masterLinkOffset));
if(masterLinkVal==0 ){
return 0;
}
long currSize = masterLinkVal>>>48;
final long pageOffset = masterLinkVal&MOFFSET;
//read packed link from stack
long ret = vol.getLongPackBidiReverse(pageOffset+currSize);
//extract number of read bytes
long oldCurrSize = currSize;
currSize-= ret >>>56;
//clear bytes occupied by prev value
vol.clear(pageOffset+currSize, pageOffset+oldCurrSize);
//and finally set return value
ret = longParityGet(ret & DataIO.PACK_LONG_RESULT_MASK);
if(CC.ASSERT && currSize<8)
throw new DBException.DataCorruption();
//is there space left on current page?
if(currSize>8){
//yes, just update master link
headVol.putLong(masterLinkOffset, parity4Set(currSize << 48 | pageOffset));
return ret;
}
//there is no space at current page, so delete current page and update master pointer
long prevPageOffset = parity4Get(vol.getLong(pageOffset));
final int currPageSize = (int) (prevPageOffset>>>48);
prevPageOffset &= MOFFSET;
//does previous page exists?
if(prevPageOffset!=0) {
//yes previous page exists
//find pointer to end of previous page
// (data are packed with var size, traverse from end of page, until zeros
//first read size of current page
currSize = parity4Get(vol.getLong(prevPageOffset)) >>> 48;
//now read bytes from end of page, until they are zeros
while (vol.getUnsignedByte(prevPageOffset + currSize-1) == 0) {
currSize--;
}
if (CC.ASSERT && currSize < 10)
throw new DBException.DataCorruption();
}else{
//no prev page does not exist
currSize=0;
}
//update master link with curr page size and offset
headVol.putLong(masterLinkOffset, parity4Set(currSize<<48 | prevPageOffset));
//release old page, size is stored as part of prev page value
freeDataPut(pageOffset, currPageSize);
return ret;
}
protected long longStackCount(final long masterLinkOffset){
if(CC.ASSERT && !structuralLock.isHeldByCurrentThread())
throw new AssertionError();
if (CC.ASSERT && (masterLinkOffset <= 0 || masterLinkOffset > PAGE_SIZE || masterLinkOffset % 8 != 0))
throw new DBException.DataCorruption("wrong master link");
long nextLinkVal = DataIO.parity4Get(
headVol.getLong(masterLinkOffset));
long ret = 0;
while(true){
long currSize = nextLinkVal>>>48;
final long pageOffset = nextLinkVal&MOFFSET;
if(pageOffset==0)
break;
//now read bytes from end of page, until they are zeros
while (vol.getUnsignedByte(pageOffset + currSize-1) == 0) {
currSize--;
}
//iterate from end of page until start of page is reached
while(currSize>8){
long read = vol.getLongPackBidiReverse(pageOffset+currSize);
//extract number of read bytes
currSize-= read >>>56;
ret++;
}
nextLinkVal = DataIO.parity4Get(
vol.getLong(pageOffset));
}
return ret;
}
@Override
public void close() {
if(closed==true)
return;
commitLock.lock();
try {
if(closed==true)
return;
flush();
vol.close();
vol = null;
if(this instanceof StoreCached)
headVol.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() {
commitLock.lock();
try {
flush();
}finally{
commitLock.unlock();
}
}
protected void flush() {
if(isReadOnly())
return;
structuralLock.lock();
try{
headVol.putLong(LAST_PHYS_ALLOCATED_DATA_OFFSET, parity3Set(lastAllocatedData));
//and set header checksum
vol.putInt(HEAD_CHECKSUM, headChecksum(vol));
}finally {
structuralLock.unlock();
}
vol.sync();
}
@Override
public void rollback() throws UnsupportedOperationException {
throw new UnsupportedOperationException();
}
@Override
public boolean canRollback() {
return false;
}
@Override
public Engine snapshot() throws UnsupportedOperationException {
if(!snapshotEnable)
throw new UnsupportedOperationException();
return new Snapshot(StoreDirect.this);
}
@Override
public void clearCache() {
}
@Override
public void backup(OutputStream out, boolean incremental) {
//lock everything
for(ReadWriteLock lock:locks){
lock.writeLock().lock();
}
try {
long maxRecid = DataIO.parity1Get(headVol.getLong(MAX_RECID_OFFSET)) / indexValSize;
recidLoop:
for (long recid = 1; recid <= maxRecid; recid++) {
long indexOffset = recidToOffset(recid);
long indexVal = vol.getLong(indexOffset);
if(checksum &&
vol.getUnsignedShort(indexOffset+8)!=
(DataIO.longHash(indexVal)&0xFFFF)){
throw new DBException.ChecksumBroken();
}
//check if was discarded
if((indexVal&MUNUSED)!=0||indexVal == 0){
continue recidLoop;
}
//check if recid was modified since last incrementa thingy
if(incremental && (indexVal&MARCHIVE)==0){
continue recidLoop;
}
//TODO we need write lock to do this, there could be setting make backup without archive marker, but only under readlock
//mark value as not modified
indexVal = DataIO.parity1Get(indexVal);
indexValPut(recid, (int) (indexVal>>>48), indexVal&MOFFSET,
(indexVal&MLINKED)!=0, false);
//write recid
DataIO.packLong(out, recid);
//load record
long[] offsets = offsetsGet(lockPos(recid),indexVal);
int totalSize = offsetsTotalSize(offsets);
if(offsets!=null) {
byte[] b = getLoadLinkedRecord(offsets, totalSize);
//write size and data
DataIO.packLong(out, b.length+1);
out.write(b);
}else{
DataIO.packLong(out, 0);
}
//TODO checksums
}
//EOF mark
DataIO.packLong(out,-1);
}catch (IOException e){
throw new DBException.VolumeIOError(e);
}finally {
//unlock everything in reverse order to prevent deadlocks
for(int i=locks.length-1;i>=0;i--){
locks[i].writeLock().unlock();
}
}
}
@Override
public void backupRestore(InputStream[] ins) {
//check we are empty
if(RECID_LAST_RESERVED+1!=DataIO.parity1Get(headVol.getLong(MAX_RECID_OFFSET))/indexValSize){
throw new DBException.WrongConfig("Can not restore backup, this store is not empty!");
}
for(ReadWriteLock lock:locks){
lock.writeLock().lock();
}
structuralLock.lock();
try {
BitSet usedRecid = new BitSet();
streamsLoop:
for(int i=ins.length-1;i>=0;i--) {
InputStream in = ins[i];
recidLoop:
for (; ; ) {
long recid = DataIO.unpackLong(in);
if (recid == -1) { // EOF
continue streamsLoop;
}
long len = DataIO.unpackLong(in);
if(ins.length!=1) {
if(recid>Integer.MAX_VALUE)
throw new AssertionError(); //TODO support bigger recids
if (usedRecid.get((int) recid)) {
//recid was already addressed in other incremental backup
//so skip length and continue
long toSkip = len - 1;
if (toSkip > 0) {
in.skip(toSkip);
}
continue recidLoop;
}
usedRecid.set((int) recid);
}
if (len == 0) {
//null record
indexValPut(recid, 0, 0, true, false);
} else {
byte[] data = new byte[(int) (len - 1)];
DataIO.readFully(in, data);
long[] newOffsets = freeDataTake(data.length);
pageIndexEnsurePageForRecidAllocated(recid);
putData(recid, newOffsets, data, data.length);
}
}
}
}catch (IOException e){
throw new DBException.VolumeIOError(e);
}finally {
structuralLock.unlock();
//unlock everything in reverse order to prevent deadlocks
for(int i=locks.length-1;i>=0;i--){
locks[i].writeLock().unlock();
}
}
}
@Override
public void compact() {
//check for some file used during compaction, if those exists, refuse to compact
if(compactOldFilesExists()){
return;
}
final boolean isStoreCached = this instanceof StoreCached;
commitLock.lock();
try{
for(int i=0;i=0;i--) {
Lock lock = locks[i].writeLock();
lock.unlock();
}
}
}finally{
commitLock.unlock();
}
}
protected boolean compactOldFilesExists() {
if(fileName!=null){
for(String s:new String[]{".compact_orig",".compact",".wal.c" ,".wal.c.compact" }) {
File oldData = new File(fileName + s);
if (oldData.exists()) {
LOG.warning("Old compaction data exists, compaction not started: " + oldData);
return true;
}
}
}
return false;
}
protected void snapshotCloseAllOnCompact() {
//close all snapshots
if(snapshotEnable){
boolean someClosed = false;
for(Snapshot snap:snapshots){
someClosed = true;
snap.close();
}
if(someClosed)
LOG.log(Level.WARNING, "Compaction closed existing snapshots.");
}
}
protected void compactIndexPages(final StoreDirect target, final AtomicLong maxRecid) {
int lastIndexPage = indexPages.length;
// make maxRecid lower if possible
// decrement maxRecid until non-empty recid is found
recidLoop: for(;;){
if(maxRecid.get()<=RECID_LAST_RESERVED){
//some recids are reserved, so break if we reach those
break recidLoop;
}
long indexVal = indexValGetRaw(maxRecid.get());
if ((indexVal & MUNUSED) == 0 && indexVal != 0) {
//non empty recid found, so break this loop
break recidLoop;
}
// maxRecid is empty, so decrement and move on
maxRecid.decrementAndGet();
}
//iterate over index pages
long maxRecidOffset = recidToOffset(maxRecid.get());
if(executor == null) {
for (int indexPageI = 0;
indexPageI < lastIndexPage && indexPages[indexPageI]<=maxRecidOffset;
indexPageI++) {
compactIndexPage(target, indexPageI, maxRecid.get());
}
}else {
//compact pages in multiple threads.
//there are N tasks (index pages) running in parallel.
//main thread checks number of tasks in interval, if one is finished it will
//schedule next one
final List tasks = new ArrayList();
for (int indexPageI = 0;
indexPageI < lastIndexPage && indexPages[indexPageI]<=maxRecidOffset;
indexPageI++) {
final int indexPageI2 = indexPageI;
//now submit tasks to executor, it will compact single page
//TODO handle RejectedExecutionException?
Future f = executor.submit(new Runnable() {
@Override
public void run() {
compactIndexPage(target, indexPageI2, maxRecid.get());
}
});
tasks.add(f);
}
//all index pages are running or were scheduled
//wait for all index pages to finish
for(Future f:tasks){
try {
f.get();
} catch (InterruptedException e) {
throw new DBException.Interrupted(e);
} catch (ExecutionException e) {
//TODO check cause and rewrap it
throw new RuntimeException(e);
}
}
}
}
protected void compactIndexPage(StoreDirect target, int indexPageI, long maxRecid) {
final long indexPage = indexPages[indexPageI];
long recid = (indexPageI==0? 0 : indexPageI * (PAGE_SIZE-8)/indexValSize - HEAD_END/indexValSize);
final long indexPageStart = (indexPage==0?HEAD_END+8 : indexPage+8);
final long indexPageEnd = indexPage+PAGE_SIZE;
//iterate over indexOffset values
//TODO check if preloading and caching of all indexVals on this index page would improve performance
indexVal:
for( long indexOffset=indexPageStart;
indexOffsetmaxRecid)
break indexVal;
final long indexVal = vol.getLong(indexOffset);
if(checksum &&
vol.getUnsignedShort(indexOffset+8)!=
(DataIO.longHash(indexVal)&0xFFFF)){
throw new DBException.ChecksumBroken();
}
//check if was discarded
if((indexVal&MUNUSED)!=0||indexVal == 0){
//mark rec id as free, so it can be reused
target.structuralLock.lock();
target.longStackPut(FREE_RECID_STACK, recid, false);
target.structuralLock.unlock();
continue indexVal;
}
//deal with linked record non zero record
if((indexVal & MLINKED)!=0 && indexVal>>>48!=0){
//load entire linked record into byte[]
long[] offsets = offsetsGet(lockPos(recid),indexValGet(recid));
int totalSize = offsetsTotalSize(offsets);
byte[] b = getLoadLinkedRecord(offsets, totalSize);
//now put into new store, acquire locks
target.locks[lockPos(recid)].writeLock().lock();
target.structuralLock.lock();
//allocate space
long[] newOffsets = target.freeDataTake(totalSize);
target.pageIndexEnsurePageForRecidAllocated(recid);
target.putData(recid,newOffsets,b, totalSize);
target.structuralLock.unlock();
target.locks[lockPos(recid)].writeLock().unlock();
continue indexVal;
}
target.locks[lockPos(recid)].writeLock().lock();
target.structuralLock.lock();
target.pageIndexEnsurePageForRecidAllocated(recid);
//TODO preserver archive flag
target.updateFromCompact(recid, indexVal, vol);
target.structuralLock.unlock();
target.locks[lockPos(recid)].writeLock().unlock();
}
}
private void updateFromCompact(long recid, long indexVal, Volume oldVol) {
//allocate new space
int size = (int) (indexVal>>>48);
long newOffset[];
if(size>0) {
newOffset=freeDataTake(size);
if (newOffset.length != 1)
throw new DBException.DataCorruption();
//transfer data
oldVol.transferInto(indexVal & MOFFSET, this.vol, newOffset[0]&MOFFSET, size);
}else{
newOffset = new long[1];
}
//update index val
//TODO preserver archive flag
indexValPut(recid, size, newOffset[0]&MOFFSET, (indexVal&MLINKED)!=0, false);
}
protected long indexValGet(long recid) {
if(CC.ASSERT)
assertReadLocked(recid);
long offset = recidToOffset(recid);
long indexVal = vol.getLong(offset);
if(indexVal == 0)
throw new DBException.EngineGetVoid();
if(checksum){
int checksum = vol.getUnsignedShort(offset+8);
if(checksum!=(DataIO.longHash(indexVal)&0xFFFF)){
throw new DBException.ChecksumBroken();
}
}
//check parity and throw recid does not exist if broken
return DataIO.parity1Get(indexVal);
}
protected long indexValGetRaw(long recid) {
if(CC.ASSERT)
assertReadLocked(recid);
long offset = recidToOffset(recid);
return vol.getLong(offset);
}
protected final long recidToOffset(long recid){
if(CC.ASSERT && recid<=0)
throw new DBException.DataCorruption("negative recid: "+recid);
if(checksum){
return recidToOffsetChecksum(recid);
}
//convert recid to offset
recid = (recid-1) * indexValSize + HEAD_END + 8;
recid+= Math.min(1, recid/PAGE_SIZE)* //if(recid>=PAGE_SIZE)
(8 + ((recid-PAGE_SIZE)/(PAGE_SIZE-8))*8);
//look up real offset
recid = indexPages[((int) (recid / PAGE_SIZE))] + recid%PAGE_SIZE;
return recid;
}
private long recidToOffsetChecksum(long recid) {
//convert recid to offset
recid = (recid-1) * indexValSize + HEAD_END + 8;
if(recid+ indexValSize >PAGE_SIZE){
//align from zero page
recid+=2+8;
}
//align for every other page
//TODO optimize away loop
for(long page=PAGE_SIZE*2;recid+ indexValSize >page;page+=PAGE_SIZE){
recid+=8+(PAGE_SIZE-8)% indexValSize;
}
//look up real offset
recid = indexPages[((int) (recid / PAGE_SIZE))] + recid%PAGE_SIZE;
return recid;
}
/** check if recid offset fits into current allocated structure */
protected boolean recidTooLarge(long recid) {
try{
recidToOffset(recid);
return false;
}catch(ArrayIndexOutOfBoundsException e){
//TODO hack
return true;
}
}
protected static long composeIndexVal(int size, long offset,
boolean linked, boolean unused, boolean archive){
if(CC.ASSERT && (size&0xFFFF)!=size)
throw new DBException.DataCorruption("size too large");
if(CC.ASSERT && (offset&MOFFSET)!=offset)
throw new DBException.DataCorruption("offset too large");
offset = (((long)size)<<48) |
offset |
(linked?MLINKED:0L)|
(unused?MUNUSED:0L)|
(archive?MARCHIVE:0L);
return parity1Set(offset);
}
/** returns new recid, recid slot is allocated and ready to use */
protected long freeRecidTake() {
if(CC.ASSERT && !structuralLock.isHeldByCurrentThread())
throw new AssertionError();
//try to reuse recid from free list
long currentRecid = longStackTake(FREE_RECID_STACK,false);
if(currentRecid!=0) {
return currentRecid;
}
currentRecid = parity1Get(headVol.getLong(MAX_RECID_OFFSET));
currentRecid+=indexValSize;
headVol.putLong(MAX_RECID_OFFSET, parity1Set(currentRecid));
currentRecid/=indexValSize;
//check if new index page has to be allocated
if(recidTooLarge(currentRecid)){
pageIndexExtend();
}
return currentRecid;
}
protected void indexLongPut(long offset, long val){
vol.putLong(offset,val);
}
protected void pageIndexEnsurePageForRecidAllocated(long recid) {
if(CC.ASSERT && !structuralLock.isHeldByCurrentThread())
throw new AssertionError();
//convert recid into Index Page number
//TODO is this correct?
recid = recid * indexValSize + HEAD_END;
recid = recid / (PAGE_SIZE-8);
while(indexPages.length<=recid)
pageIndexExtend();
}
protected void pageIndexExtend() {
if(CC.ASSERT && !structuralLock.isHeldByCurrentThread())
throw new AssertionError();
//allocate new index page
long indexPage = pageAllocate();
//add link to previous page
long nextPagePointerOffset = indexPages[indexPages.length-1];
//if zero page, put offset to end of page
nextPagePointerOffset = Math.max(nextPagePointerOffset, HEAD_END);
indexLongPut(nextPagePointerOffset, parity16Set(indexPage));
//set zero link on next page
indexLongPut(indexPage,parity16Set(0));
//put into index page array
long[] indexPages2 = Arrays.copyOf(indexPages,indexPages.length+1);
indexPages2[indexPages.length]=indexPage;
indexPages = indexPages2;
}
protected long pageAllocate() {
if(CC.ASSERT && !structuralLock.isHeldByCurrentThread())
throw new AssertionError();
long storeSize = parity16Get(headVol.getLong(STORE_SIZE));
vol.ensureAvailable(storeSize+PAGE_SIZE);
vol.clear(storeSize,storeSize+PAGE_SIZE);
headVol.putLong(STORE_SIZE, parity16Set(storeSize + PAGE_SIZE));
if(CC.ASSERT && storeSize%PAGE_SIZE!=0)
throw new DBException.DataCorruption();
return storeSize;
}
protected static int round16Up(int pos) {
//TODO optimize this, no conditions
int rem = pos&15; // modulo 16
if(rem!=0) pos +=16-rem;
return pos;
}
public static final class Snapshot extends ReadOnly{
protected StoreDirect engine;
protected LongLongMap[] oldRecids;
public Snapshot(StoreDirect engine){
this.engine = engine;
oldRecids = new LongLongMap[engine.lockScale];
for(int i=0;i A get(long recid, Serializer serializer) {
StoreDirect engine = this.engine;
int pos = engine.lockPos(recid);
Lock lock = engine.locks[pos].readLock();
lock.lock();
try{
long indexVal = oldRecids[pos].get(recid);
if(indexVal==-1)
return null; //null or deleted object
if(indexVal==-2)
return null; //TODO deserialize empty object
if(indexVal!=0){
long[] offsets = engine.offsetsGet(pos, indexVal);
return engine.getFromOffset(serializer,offsets);
}
return engine.get2(recid,serializer);
}finally {
lock.unlock();
}
}
@Override
public void close() {
//TODO lock here?
engine.snapshots.remove(Snapshot.this);
engine = null;
oldRecids = null;
//TODO put oldRecids into free space
}
@Override
public boolean isClosed() {
return engine!=null;
}
@Override
public boolean canRollback() {
return false;
}
@Override
public boolean canSnapshot() {
return true;
}
@Override
public Engine snapshot() throws UnsupportedOperationException {
return this;
}
@Override
public Engine getWrappedEngine() {
return engine;
}
@Override
public void clearCache() {
}
}
}
Ads help maintain this website.