/** 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<<CC.VOLUME_PAGE_SHIFT;

protected static final long MOFFSET = 0x0000FFFFFFFFFFF0L;

protected static final long MLINKED = 0x8L;

protected static final long MUNUSED = 0x4L;

protected static final long MARCHIVE = 0x2L;

protected static final long MPARITY = 0x1L;

protected static final long STORE_SIZE = 8*2;

/** physical offset of maximal allocated recid. Parity1.

protected static final long MAX_RECID_OFFSET = 8*3;

protected static final long LAST_PHYS_ALLOCATED_DATA_OFFSET = 8*4; //TODO update doc

protected static final long FREE_RECID_STACK = 8*5;

/*following slots might be used in future */

protected static final long UNUSED1 = 8*6;

protected static final long UNUSED2 = 8*7;

protected static final long UNUSED3 = 8*8;

protected static final long UNUSED4 = 8*9;

protected static final long UNUSED5 = 8*10;

protected static final int MAX_REC_SIZE = 0xFFFF;

/** number of free physical slots */

protected static final int SLOTS_COUNT = 2+(MAX_REC_SIZE)/16; //it rounds down, plus extra slot for zeros (not really used)

protected static final long HEAD_END = UNUSED5 + SLOTS_COUNT * 8;

protected static final long INITCRC_INDEX_PAGE = 4329042389490239043L;

protected static final long[] EMPTY_LONGS = new long[0];

//TODO this refs are swapped during compaction. Investigate performance implications

protected volatile Volume vol;

protected volatile Volume headVol;

//TODO this only grows under structural lock, but reads are outside structural lock, does it have to be volatile?

protected volatile long[] indexPages;

protected volatile long lastAllocatedData=0; //TODO this is under structural lock, does it have to be volatile?

protected final ScheduledExecutorService executor;

protected final List<Snapshot> 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<Snapshot>():

null;

this.indexValSize = checksum ? 10 : 8;

this.sizeIncrement = Math.max(1L<<CC.VOLUME_PAGE_SHIFT, DataIO.nextPowTwo(sizeIncrement));

this.startSize = Fun.roundUp(Math.max(1L<<CC.VOLUME_PAGE_SHIFT,startSize), this.sizeIncrement);

this.recidReuseDisable = recidReuseDisable;

this.sliceShift = Volume.sliceShiftFromSize(this.sizeIncrement);

if(CC.LOG_STORE && LOG.isLoggable(Level.FINE)){

LOG.log(Level.FINE, "StoreDirect constructed: executor={0}, snapshots={1}, indexValSize={2}, " +

"startSize={3}, sizeIncrement={4}",

new Object[]{executor,snapshots,indexValSize,startSize,sizeIncrement});

}

}

@Override

public void init() {

commitLock.lock();

try {

structuralLock.lock();

try {

boolean empty = Volume.isEmptyFile(fileName);

this.vol = volumeFactory.makeVolume(fileName, readonly, fileLockDisable, sliceShift, startSize, false);

if (empty) {

initCreate();

} else {

initOpen();

}

} finally {

structuralLock.unlock();

}

}catch(RuntimeException e){

initFailedCloseFiles();

if(vol!=null && !vol.isClosed()) {

vol.close();

}

vol = null;

throw e;

}finally {

commitLock.unlock();

}

}

protected void initFailedCloseFiles() {

}

protected void initOpen() {

if(CC.ASSERT && !commitLock.isHeldByCurrentThread())

throw new AssertionError();

if(CC.ASSERT && !structuralLock.isHeldByCurrentThread())

throw new AssertionError();

int header = vol.getInt(0);

if(header!=header){

throw new DBException.WrongConfig("This is not MapDB file");

}

if (CC.LOG_STORE && LOG.isLoggable(Level.FINE)) {

LOG.log(Level.FINE, "initOpen: file={0}, volLength={1}, vol={2}", new Object[]{fileName, vol.length(), vol});

}

if(vol.getInt(0)!=HEADER){

//TODO handle version numbers

throw new DBException.DataCorruption("wrong header in file: "+fileName);

}

//check header config

checkFeaturesBitmap(vol.getLong(HEAD_FEATURES));

initHeadVol();

//check head checksum

int expectedChecksum = vol.getInt(HEAD_CHECKSUM);

int actualChecksum = headChecksum(vol);

if (actualChecksum != expectedChecksum) {

throw new DBException.HeadChecksumBroken();

}

//load index pages

long[] ip = new long[]{0};

long indexPage = parity16Get(vol.getLong(HEAD_END));

int i=1;

for(;indexPage!=0;i++){

if(CC.ASSERT && indexPage%PAGE_SIZE!=0)

throw new DBException.DataCorruption();

if(ip.length==i){

ip = Arrays.copyOf(ip, ip.length * 4);

}

ip[i] = indexPage;

//move to next page

indexPage = parity16Get(vol.getLong(indexPage));

}

indexPages = Arrays.copyOf(ip,i);

lastAllocatedData = parity3Get(vol.getLong(LAST_PHYS_ALLOCATED_DATA_OFFSET));

if (CC.LOG_STORE && LOG.isLoggable(Level.FINEST)) {

LOG.log(Level.FINEST, "indexPages: {0}", Arrays.toString(indexPages));

LOG.log(Level.FINEST, "lastAllocatedData: {0}", lastAllocatedData);

}

}

protected void initCreate() {

if(CC.ASSERT && !commitLock.isHeldByCurrentThread())

throw new AssertionError();

if(CC.ASSERT && !structuralLock.isHeldByCurrentThread())

throw new AssertionError();

//create initial structure

//set features bitmap

final long features = makeFeaturesBitmap();

if (CC.LOG_STORE && LOG.isLoggable(Level.FINE)) {

LOG.log(Level.FINE, "initCreate: file={0}, volLength={1}, vol={2}, features={3}",

new Object[]{fileName, vol.length(), vol, Long.toBinaryString(features)});

}

//create new store

indexPages = new long[]{0};

vol.ensureAvailable(PAGE_SIZE);

vol.clear(0, PAGE_SIZE);

//set sizes

vol.putLong(STORE_SIZE, parity16Set(PAGE_SIZE));

vol.putLong(MAX_RECID_OFFSET, parity1Set(RECID_LAST_RESERVED * indexValSize));

//pointer to next index page (zero)

vol.putLong(HEAD_END, parity16Set(0));

lastAllocatedData = 0L;

vol.putLong(LAST_PHYS_ALLOCATED_DATA_OFFSET, parity3Set(lastAllocatedData));

//put reserved recids

for(long recid=1;recid<RECID_FIRST;recid++){

long indexVal = parity1Set(MLINKED | MARCHIVE);

long indexOffset = recidToOffset(recid);

vol.putLong(indexOffset, indexVal);

if(checksum) {

vol.putUnsignedShort(indexOffset + 8, DataIO.longHash(indexVal)&0xFFFF);

}

}

//put long stack master links

for(long masterLinkOffset = FREE_RECID_STACK;masterLinkOffset<HEAD_END;masterLinkOffset+=8){

vol.putLong(masterLinkOffset,parity4Set(0));

}

//write header

vol.putInt(0,HEADER);

vol.putLong(HEAD_FEATURES, features);

//and set header checksum

vol.putInt(HEAD_CHECKSUM, headChecksum(vol));

vol.sync();

initHeadVol();

}

protected void initHeadVol() {

if(CC.ASSERT && !structuralLock.isHeldByCurrentThread())

throw new AssertionError();

this.headVol = vol;

}

public StoreDirect(String fileName) {

this(fileName,

fileName==null? CC.DEFAULT_MEMORY_VOLUME_FACTORY : CC.DEFAULT_FILE_VOLUME_FACTORY,

null,

CC.DEFAULT_LOCK_SCALE,

0,

false,false,null,false,false,false,null,

null, 0L, 0L, false);

}

protected int headChecksum(Volume vol2) {

if(CC.ASSERT && !structuralLock.isHeldByCurrentThread())

throw new AssertionError();

int ret = 0;

for(int offset = 8;

offset< HEAD_END;

offset+=8){

long val = vol2.getLong(offset);

ret += DataIO.longHash(offset + val);

}

if (CC.LOG_STORE && LOG.isLoggable(Level.FINEST)) {

LOG.log(Level.FINEST, "headChecksum={0}", Integer.toHexString(ret));

}

return ret;

}

@Override

protected <A> A get2(long recid, Serializer<A> serializer) {

if (CC.ASSERT)

assertReadLocked(recid);

long[] offsets = offsetsGet(lockPos(recid),indexValGet(recid));

return getFromOffset(serializer, offsets);

}

protected <A> A getFromOffset(Serializer<A> 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<ret.length;i++) {

boolean last = (i==ret.length-1);

boolean linked = (ret[i]&MLINKED)!=0;

if(!last && !linked)

throw new DBException.DataCorruption("body not linked");

if(last && linked)

throw new DBException.DataCorruption("tail is linked");

long offset = ret[i]&MOFFSET;

if(offset<PAGE_SIZE)

throw new DBException.DataCorruption("offset is too small");

if(((offset&MOFFSET)%16)!=0)

throw new DBException.DataCorruption("offset not mod 16");

int size = (int) (ret[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 <A> void delete2(long recid, Serializer<A> 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();

}

}

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 <A> long put(A value, Serializer<A> 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<PAGE_SIZE))

throw new DBException.DataCorruption("wrong offset");

if (CC.LOG_STORE && LOG.isLoggable(Level.FINEST)) {

LOG.log(Level.FINEST, "offset={0}, size={1}",

new Object[]{offset, size});

}

if(!(this instanceof StoreWAL)) //TODO WAL needs to handle record clear, perhaps WAL instruction?

vol.clear(offset,offset+size);

//shrink store if this is last record

if(offset+size==lastAllocatedData){

lastAllocatedData-=size;

return;

}

freeSizeIncrement(size);

long masterPointerOffset = size/2 + FREE_RECID_STACK; // really is size*8/16

longStackPut(

masterPointerOffset,

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 && ret<PAGE_SIZE)

throw new DBException.DataCorruption("wrong ret");

if(CC.ASSERT && ret%16!=0)

throw new DBException.DataCorruption("unalligned ret");

if (CC.LOG_STORE && LOG.isLoggable(Level.FINEST)) {

LOG.log(Level.FINEST, "size={0}, ret={1}",

new Object[]{size, Long.toHexString(ret)});

}

freeSizeIncrement(-size);

return ret;

}

if(lastAllocatedData==0){

//allocate new data page

long page = pageAllocate();

lastAllocatedData = page+size;

if(CC.ASSERT && page<PAGE_SIZE)

throw new DBException.DataCorruption("wrong page");

if(CC.ASSERT && page%16!=0)

throw new DBException.DataCorruption("unalligned page");

if (CC.LOG_STORE && LOG.isLoggable(Level.FINEST)) {

LOG.log(Level.FINEST, "size={0}, ret={1}",

new Object[]{size, Long.toHexString(ret)});

}

return page;

}

//does record fit into rest of the page?

if((lastAllocatedData%PAGE_SIZE + size)/PAGE_SIZE !=0){

//throw away rest of the page and allocate new

lastAllocatedData=0;

freeDataTakeSingle(size);

//TODO i thing return! should be here, but not sure.

//TODO it could be possible to recycle data here.

// save pointers and put them into free list after new page was allocated.

}

//yes it fits here, increase pointer

ret = lastAllocatedData;

lastAllocatedData+=size;

if(CC.ASSERT && ret%16!=0)

throw new DBException.DataCorruption();

if(CC.ASSERT && lastAllocatedData%16!=0)

throw new DBException.DataCorruption();

if(CC.ASSERT && ret<PAGE_SIZE)

throw new DBException.DataCorruption();

if (CC.LOG_STORE && LOG.isLoggable(Level.FINEST)) {

LOG.log(Level.FINEST, "size={0}, ret={1}",

new Object[]{size, Long.toHexString(ret)});

}

return ret;

}

//TODO use var size

protected final static long CHUNKSIZE = 100*16;

protected void longStackPut(final long masterLinkOffset, final long value, boolean recursive){

if(CC.ASSERT && !structuralLock.isHeldByCurrentThread())

throw new AssertionError();

if(CC.ASSERT && (masterLinkOffset<=0 || masterLinkOffset>PAGE_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 && (masterLinkOffset<FREE_RECID_STACK ||

masterLinkOffset>FREE_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))