unit uFile;

{$DEFINE SAFETY} //extra sense checks, e.g. page=0

uses uPage, uGlobal, uStmt, uGlobalDef, IdTCPConnection{debug only};

TFileDirSize=sizeof(pageId)+sizeof(word) +2; //Note: +2 because record seems to round up

FileDirPerBlock=BlockSize div TFileDirSize;

InvalidDirSlot=-1;

TfileDir=record //note: see TFileDirSize above - keep in sync!!

pid:PageId;

Space:word; //(limits blockSize to 65535)

end; {TfileDir}

recType=(rtHeader, {Special slot header}

rtEmpty, {Slot no longer used - can be purged if at end of page slot array}

rtDeletedRecord, {Deleted header - retain - deletion could be rolled-back => same as normal rtRecord}

rtRecord, {Record header}

rtDelta, {Record delta}

rtReservedSlot, {Note: this is for internal page shuffling only - it should never be seen on disk!}

rtBlob {Blob data - reference by rtRecord or rtDelta}

); //todo check/force size is byte? = default (or use $MINENUMSIZE 1)

//todo keep in sync with recTypeText strings below

{(These limit the max blockSize to 65535)} //note: ensure algorithms don't try to go past the limit

RecId=word; //in-page pointer

RecSize=word; //in-page record length

PtrRecData=^TrecData;

TrecData=array [0..MaxRecSize-1] of char; //record data (used for pointer typing and AddRecord tests)

//note similarity between TSlot and Trec! -same except Trec has a swizzled pointer to memory

//maybe could use Tslot.start(longint) as memory pointer? saving what?

Trec=class //Record block pointer within a (pinned) page (overlay)

private

public

rType:recType; //record type

dataPtr:PtrRecData; //data start

len:integer; //length

Wt:StampId; //Transaction W timestamp

prevRID:Trid; //pointer to previous record in db

end; {Trec}

TDBFile=class //database heapfile //todo abstract?

protected

fstartPage:PageId; //first page of page directory

fname:string; //the db filename

//todo move these to the transaction level & have scan routines pass the latest value to us

// that way multiple users can share the same file (esp. db.sysTable etc.) but can

// scan independently.

// So, make sure we make this thread-safe: esp. access to properties/open/new routines etc.

//This then allows them to share heapfile & so Trelation

// - again make sure these are thread safe:

// - ->we need to move relation.fTuple to the transaction also!

// but then iterator trees would collapse!

// we really need a new ftuple per tran using the relation

// but then there's little point sharing the relation!

// Think/design...

// only need to share ftuple (and file.currentRid) for db.sysTableRel etc.

// and these are only used for quick lookup access (don't serialise!)

// and occasionaly insertions/updates (could be serialised)

// So, each tran really needs a currentRid + currentTuple for each sysRel

// to allow concurrent scanning/reading (no sharing 1 tuple for all sysRel, since may be fast-joining etc)

// -probably will be around: col,table,auth,schema,dom,constr,dom-con,col-con,fk,idx= 10

// system relations = extra ~10k per transaction = not bad

// alternative is to keep as they are, via db = bottleneck

// but bottle neck only when

// opening/creating relations (then in tuple def)

// building constraint trees for insert/update (will be done once at start)

// so maybe hardly ever any contention????? so extra complexity not worth it?

// -maybe in future could add another thread/set of sysRels to handle extra requests in parallel

// - i.e. 1 or 2 sets of sysRels should handle it?

// We should replace hCatalogMutex with hsysTableCriticalSection, hsysColumnCriticalSection etc.

// and this means all system catalog functions must be made atomic -via db.routines? - ok?

// - may need sysCatalog.lock; sysCatalog.findNextColumn; ...; finally sysCatalog.unlock

// plus, these routines make it easier to have an extra catalog server thread later...

// plus they hide the details!

// but maybe for speed each tran should have its own sysTableRel,sysColumnRel?

// (etc. -use array of rels, e.g. tr.sysCatalog[sysTable].startscan)

// then rel.open is very fast with no bottlenecks

// - other stuff like reading constraints/permissions

// can be left to db routines (serialised, but maybe less so later...)

// use tr.sysCatalog[] array for all sys relations

// - for those we can serialise for now, point the rel at db.sysRel instead of creating & opening it!

// - i.e. looks independent but is shared resource - who serialises access to it?

// - maybe use Trelation.serialiseAccess

// and all scanstart's rel.lock which will either do nothing or grab critical section

// and all scanstop's rel.unlock -prone to error? -not very neat? waste CS per relation

// maybe db.FindSysDomain() can be used for now

// or all use shared db.FindNextCol(tr,rel)

// & if passed non-db rel, use else serialise access...

// e.g. tr: db.FindNextCol(tr,tr.sysCatalog[sysColumn])

// tr: db.FindNextDomain(tr,db.sysDomain) -> serialised cos rel.owner=db

// tr: db.FindFirstCol(tr,tr.sysCatalog[sysColumn],'schema1','table1')

// but then may as well get findFirstCol to use tr.sysCatalog[sysColumn]

// and findFirstDomain to use db.sysDomain - save params

// -still isolated for user & can be changed later by adding to

// tr.sysCatalog array & tr.start (to open sysRels) & server routines

// plus catalog access code is in db routines where it belongs (not in tuple/rel routines)

// plus allows us to serialise/centralise issuing of new id's etc.

//

fCurrentRID:Trid; //scan pointer (fCurrentRID.pid will be pinned during scan)

fCurrentPage:TPage; //scan page

fDirPageFindStartPage:PageId; //first page of page directory that we already found a space on

fDirPageFindDirPageOffset:integer; //page offset from fStartPage for fDirPageFindStartPage

fDirPageFindSpace:word; //last space sought

public

property startPage:PageId read fStartPage;

property name:string read fname; //only currently used for index name debug messages, so remove

property currentRID:Trid read fCurrentRID; //need to reference from tuple for deletions

constructor Create; virtual;

destructor Destroy; override;

function createFile(st:Tstmt;const fname:string):integer; virtual;

function deleteFile(st:Tstmt):integer; virtual;

function openFile(st:Tstmt;const filename:string;startPage:PageId):integer; virtual;

function freeSpace(st:Tstmt;page:TPage):integer; virtual;

//TODO change to ReadRecord(rid:Trid,r:Trec):integer to match AddRecord etc.!

// also move to HeapFile? i.e. make virtual;abstract - don't assume slot structure at this level? maybe?

function ReadRecord(st:Tstmt;page:TPage;sid:SlotId;r:Trec):integer; virtual; abstract;

function AddRecord(st:Tstmt;r:Trec;var rid:Trid):integer; virtual; abstract;

{File directory is used for:

function DirCount(st:Tstmt;var dirSlotCount:integer;var lastDirPage,prevLastSlotDirPage:PageId;retryForAccuracy:boolean):integer;

function DirPage(st:Tstmt;DirPageId:PageId;dirSlot:DirSlotId;var pid:PageId;var space:word):integer;

function DirPageSet(st:Tstmt;DirPageId:PageId;dirSlot:DirSlotId;pid:PageId;space:word;AllowOverwrite:boolean):integer;

function DirPageFind(st:Tstmt;space:word;var pid:PageId;var DirPageId:PageId;var dirSlot:DirSlotId):integer;

function DirPageFindFromPID(st:Tstmt;pid:PageId;var DirPageId:PageId;var dirSlot:DirSlotId):integer;

function DirPageAdd(st:Tstmt;pid:PageId;space:word;var prevDirSlotPageId,DirPageId:PageId;var dirSlot:DirSlotId):integer;

function DirPageRemove(st:Tstmt;pid:PageId;dirSlot:DirSlotId):integer;

function DirPageAdjustSpace(st:Tstmt;DirPageId:PageId;dirSlot:DirSlotId;spaceAdj:integer;var space:word):integer;

function GetScanStart(st:Tstmt;var rid:Trid):integer; virtual;

function GetScanNext(st:Tstmt;var rid:Trid;var noMore:boolean):integer; virtual;

function GetScanStop(st:Tstmt):integer; virtual;

function debugDump(st:Tstmt;connection:TIdTCPConnection;summary:boolean):integer; virtual;

end; {TDBFile}

recTypeText:array [rtHeader..rtBlob] of string = ('header',

'empty',

'deletedRecord',

'record',

'delta',

'reserved',

'blob'

);

debugFileCreate:integer=0;

debugFileDestroy:integer=0;

uLog,

{$ENDIF}

sysUtils, uServer, uTransaction, uOS {for sleep}

,uEvsHelpers

;

where='uFile';

who='';

{Retry attempts}

RETRY_DIRPAGEADD=50;

RETRY_DIRCOUNT=50; //Note: this could be doubled by retry in DirPageAdd

dirCountEmptyBackoffMin=5; //min. milliseconds (plus random) //todo make proportional to CPU speed/active threads

dirCountEmptyBackoffExtra=50; //max. milliseconds (random) //todo make proportional to CPU speed/active threads

constructor TDBFile.Create;

inc(debugFileCreate);

{$IFDEF DEBUG_LOG}

if debugFileCreate=1 then

log.add(who,where,format(' File memory size=%d',[instanceSize]),vDebugLow);

{$ENDIF}

fStartPage:=InvalidPageId;

fCurrentRID.pid:=InvalidPageId;

fCurrentRID.sid:=InvalidSlotId;

end;

destructor TDBFile.Destroy;

const routine=':destroy';

if fCurrentRID.pid<>InvalidPageId then

begin

{$IFDEF DEBUG_LOG}

log.add(who,where+routine,format('Scan of %s starting at %d is in progress - page %d will be left pinned',[fname,fStartPage,fCurrentRID.pid]),vDebugError);

{$ELSE}

;

{$ENDIF}

end;

inc(debugFileDestroy);

inherited;

end;

function TDBFile.createFile(st:Tstmt;const fname:string):integer;

const routine=':createFile';

page:Tpage;

fileDir:TFileDir;

i:DirSlotId;

needDirEntry:boolean;

//todo assert db<>nil

result:=Fail;

needDirEntry:=(fname=sysTable_file); //only need db-dir entry for sysTable

if Ttransaction(st.owner).db.addFile(st,fname,fstartPage,needDirEntry)<>ok then

result:=Fail

else

begin

{$IFDEF DEBUG_LOG}

log.add(st.who,where+routine,format('File %s created starting at page %d',[fname,fStartPage]),vDebug);

{$ENDIF}

{Reset DirPageFind cache}

fDirPageFindStartPage:=InvalidPageId;

fDirPageFindDirPageOffset:=0;

fDirPageFindSpace:=MAX_WORD;

{Create the directory page}

//note: would be neater to do this when needed i.e. use DirPageAdd() chicken&egg

with Ttransaction(st.owner).db.owner as TDBserver do

begin

if buffer.pinPage(st,fStartPage,page)<>ok then

begin

{$IFDEF DEBUG_LOG}

log.add(st.who,where+routine,'Failed pinning header page',vDebugError);

{$ENDIF}

result:=Fail;

exit;

end;

try

if page.latch(st)=ok then //note: no real need since page is local?

begin

try

page.block.pageType:=ptFileDir; //overkill, ptFiledir is the default

//speed: page.block.prevPage:=page.block.thisPage; //make linked list a circle for fast adding later

{Write zeroised page pointers}

fileDir.pid:=InvalidPageId;

fileDir.space:=0;

for i:=0 to FileDirPerBlock-1 do

begin

page.SetBlock(st,i*sizeof(fileDir),sizeof(fileDir),@fileDir);

end;

page.dirty:=True;

finally

page.unlatch(st);

end; {try}

//30/08/00 moved:page.dirty:=True;

end

else

begin

result:=fail;

exit;

end;

finally

buffer.unpinPage(st,fStartPage);

end; {try}

end; {with}

result:=ok;

end;

end; {createFile}

function TDBFile.deleteFile(st:TStmt):integer;

const routine=':deleteFile';

page,startPage:Tpage;

pid:PageId;

fileDir:TFileDir;

i:DirSlotId;

needDirEntry:boolean;

//newpid:PageId;

prevPid:PageId;

//newPage:TPage;

prevPage:TPage;

lastDirPage:PageId;

dirSlot:DirSlotId;

//todo assert db<>nil

result:=Fail;

with Ttransaction(st.owner).db.owner as TDBserver do

begin

{Move forwards through chain & check the directory pages are clear}

pid:=fStartPage;

if buffer.pinPage(st,pid,page)<>ok then

begin

{$IFDEF DEBUG_LOG}

log.add(st.who,where+routine,'Failed pinning header page',vDebugError);

{$ENDIF}

exit;

end;

lastDirPage:=pid;

try

{Check each page in the file directory chain is clear: we can guarantee at least 1 file directory page}

while pid<>InvalidPageId do

begin

{Check that this page is ok to delete}

if page.block.pageType<>ptFileDir then

begin

{$IFDEF DEBUG_LOG}

log.add(st.who,where+routine,format('File directory page (%d) is not the expected type (%d)',[pid,ord(page.block.pageType)]),vAssertion);

{$ENDIF}

exit; //abort

end;

(*

for i:=0 to FileDirPerBlock-1 do

begin

page.AsBlock(st,i*sizeof(fileDir),sizeof(fileDir),@fileDir);

if (FileDir.pid<>InvalidPageId) or (fileDir.space<>0) then

begin

{$IFDEF DEBUG_LOG}

log.add(st.who,where+routine,format('File file directory page (%d) has an allocted slot (%d) [%d:%d]',[pid,i,FileDir.pid,fileDir.space]),vAssertion);

{$ENDIF}

exit; //abort

end;

end;

//next in chain

pid:=page.block.nextPage;

if pid<>InvalidPageId then

begin

buffer.unpinPage(st,page.block.thisPage);

if buffer.pinPage(st,pid,page)<>ok then

begin

{$IFDEF DEBUG_LOG}

log.add(st.who,where+routine,'Failed pinning page',vDebugError);

{$ENDIF}

exit;

end;

lastDirPage:=pid;

end;

//speed: else assert fStartPage.block.prevPage=page.block.thisPage? (i.e. the end is what we expected)

end;

finally

buffer.unpinPage(st,page.block.thisPage);

end; {try}

{Remove the dir pages (except the first one), starting with the last one moving backwards through the chain}

(*speed:

if buffer.pinPage(st,lastDirPage,page)<>ok then

begin

{$IFDEF DEBUG_LOG}

log.add(st.who,where+routine,'Failed pinning last directory page',vDebugError);

{$ENDIF}

exit;

end;

try

while lastDirPage<>fStartPage do

begin

{De-initialise the file directory page}

{Unlink from previous dir page}

if lastDirPage=fStartPage then

begin

prevPid:=InvalidPageId; //this is the very first directory page

(*speed:

end

else

begin

prevPid:=page.block.prevPage;

if buffer.pinPage(st,prevpid,prevpage)<>ok then

begin

{$IFDEF DEBUG_LOG}

log.add(st.who,where+routine,format('Failed reading dir page''s previous page, %d',[prevpid]),vError);

{$ENDIF}

exit; //abort

end;

prevPage.latch(st);

try

(*speed:

prevPage.block.nextPage:=InvalidPageId; //sever link forwards

prevPage.dirty:=True;

finally

prevPage.Unlatch(st);

buffer.unpinPage(st,prevpid);

end; {try}

end;

//move to previous in chain

lastDirPage:=prevPid;

if lastDirPage<>InvalidPageId then

begin

buffer.unpinPage(st,page.block.thisPage);

{Remove this directory page}

//Note: these pages do not appear in the file directory itself, so no need to remove //double-check!

if Ttransaction(st.owner).db.deallocatePage(st,page.block.thisPage)<>ok then

begin

{$IFDEF DEBUG_LOG}

log.add(st.who,where+routine,'Failed deallocating file directory page',vError);

{$ENDIF}

exit; //abort

end;

{$IFDEF DEBUG_LOG}

log.add(st.who,where+routine,format('Contracted file page directory by page %d',[page.block.thisPage]),vDebug);

{$ENDIF}

if buffer.pinPage(st,lastDirPage,page)<>ok then

begin

{$IFDEF DEBUG_LOG}

log.add(st.who,where+routine,'Failed pinning page',vDebugError);

{$ENDIF}

exit;

end;

end;

end;

finally

buffer.unpinPage(st,page.block.thisPage);

end; {try}

(*speed:

{note: removing the first directory page is done by db.removeFile}

end; {with}

needDirEntry:=(fname=sysTable_file); //only need db-dir entry for sysTable

if Ttransaction(st.owner).db.removeFile(st,fname,fstartPage,needDirEntry)<>ok then

result:=Fail

else

begin

{$IFDEF DEBUG_LOG}

log.add(st.who,where+routine,format('File %s deleted starting at page %d',[fname,fStartPage]),vDebug);

{$ENDIF}

result:=ok;

end;

end; {deleteFile}

function TDBFile.openFile(st:TStmt;const filename:string;startPage:PageId):integer;

const routine=':openFile';

page:TPage;

result:=Fail;

//todo assert db<>nil

// assert file exists?

fname:=filename;

{Get the directory page}

with Ttransaction(st.owner).db.owner as TDBserver do

begin

if buffer.pinPage(st,StartPage,page)<>ok then

begin

{$IFDEF DEBUG_LOG}

log.add(st.who,where+routine,format('Failed reading start page %d of %s',[fStartPage,filename]),vDebugError);

{$ENDIF}

exit; //abort

end;

try

fstartPage:=startPage;

{$IFDEF DEBUGDETAIL}

{$IFDEF DEBUG_LOG}

log.add(st.who,where+routine,format('File %s opened starting at page %d',[filename,fStartPage]),vDebug);

{$ENDIF}

{$ENDIF}

//todo sanity checks?

{Reset DirPageFind cache}

fDirPageFindStartPage:=InvalidPageId;

fDirPageFindDirPageOffset:=0;

fDirPageFindSpace:=MAX_WORD;

result:=ok;

finally

buffer.unpinPage(st,StartPage); //todo leave pinned until ScanNext

end; {try}

end; {with}

end; {openFile}

function TDBFile.freeSpace(st:TStmt;page:TPage):integer;

const routine=':freeSpace';

result:=BlockSize;

end; {FreeSpace}

function TDBFile.DirCount(st:TStmt;var dirSlotCount:integer;var lastDirPage,prevLastSlotDirPage:PageId;retryForAccuracy:boolean):integer;

const routine=':DirCount';

page:TPage;

fileDir:TfileDir;

dirPageOffset:integer;

i:DirSlotId;

retry:integer;

result:=Fail;

if fStartPage=InvalidPageId then

begin

{$IFDEF DEBUG_LOG}

log.add(st.who,where+routine,'Uninitialised start page',vDebugError);

{$ENDIF}

exit;

end;

with Ttransaction(st.owner).db.owner as TDBserver do

begin

retry:=-1; //i.e. don't retry at all, unless certain circumstance

while (result<>ok) and ((retry>0) or (retry=-1)) do //outer loop to retry in case we find our last dir page has a next page when we've read it

begin

if retry>0 then

begin

dec(retry); //i.e. once retry fired, no more to avoid infinite loop

{$IFDEF DEBUG_LOG}

log.add(st.who,where+routine,format('Retrying (%d more tries to go)',[retry]),vDebugMedium);

{$ENDIF}

end;

dirPageOffset:=0;

{Move to last page}

lastDirPage:=fStartPage;

if buffer.pinPage(st,lastDirPage,page)<>ok then

begin

{$IFDEF DEBUG_LOG}

log.add(st.who,where+routine,format('Failed reading next first dir page %d',[lastDirPage]),vDebugError);

{$ENDIF}

if retry=-1 then retry:=1; //todo any point?

{$IFDEF DEBUG_LOG}

log.add(st.who,where+routine,format('Retry set to %d',[retry]),vDebugLow);

{$ENDIF}

//exit;

continue; //abort/retry

end;

try

{speed: jump straight to last page rather than reading whole chain but:

while page.block.nextPage<>InvalidPageId do

begin

lastDirPage:=page.block.nextPage;

buffer.unpinPage(st,page.block.thisPage{was lastDirPage & was before lastDirPage was reset});

if buffer.pinPage(st,lastDirPage,page)<>ok then //note: use outer finally to unpin

begin

{$IFDEF DEBUG_LOG}

log.add(st.who,where+routine,format('Failed reading next dir page %d',[lastDirPage]),vDebugError);

{$ENDIF}

if retry=-1 then retry:=1; //todo any point?

{$IFDEF DEBUG_LOG}

log.add(st.who,where+routine,format('Retry set to %d',[retry]),vDebugLow);

{$ENDIF}

//exit; //abort - note will cause fail on finally, unpin

continue; //abort/retry

end;

inc(dirPageOffset);

end; {while}

{Initial count based on number of full pages read

dirSlotCount:=dirPageOffset*FileDirPerBlock;

{Now count the entries in this last page}

i:=0;

{We latch here in an attempt to make more stable (reduce retries) for multi-threaded adding}

//todo check this doesn't cause it to be slower...

if retryForAccuracy then

page.latch(st);

try

//todo check if page.block.nextPage<>InvalidPageId=>retry now to avoid delay? speed

page.AsBlock(st,i*sizeof(fileDir),sizeof(fileDir),@fileDir);

while filedir.pid<>InvalidPageId do

begin

inc(dirSlotCount);

inc(i);

if i=FileDirPerBlock then

begin //reached last entry in dir page - read next?

if page.block.nextPage<>InvalidPageId then

begin

{Note: this could be because another thread is racing away adding entries...so we retry if caller needs accuracy}

{$IFDEF DEBUG_LOG}

log.add(st.who,where+routine,format('Last dir page (%d) has a next pointer, but the initial bulk count reported none',[lastDirPage]),vDebugWarning); //i.e. race bug, e.g. during DirPageAdd! = disaster? but we retry...

{$ENDIF}

if retry=-1 then

if retryForAccuracy then

retry:=RETRY_DIRCOUNT

else

retry:=1; //todo any point? see code when i=0 below: surely this is accurate enough //speed

{$IFDEF DEBUG_LOG}

log.add(st.who,where+routine,format('Retry set to %d',[retry]),vDebugLow);

{$ENDIF}

sleepOS(dirCountEmptyBackoffMin+random(dirCountEmptyBackoffExtra));

//exit; //abort

break; //abort/retry

end;

//this last one is full, end loop

fileDir.pid:=InvalidPageId;

end;

{Look at next slot}

if fileDir.pid<>InvalidPageId then

page.AsBlock(st,i*sizeof(fileDir),sizeof(fileDir),@fileDir);

end; {while}

finally

if retryForAccuracy then

page.unlatch(st);

end; {try}

if filedir.pid<>InvalidPageId then continue; //must have broken out in error so jump to retry

if (dirPageOffset>0) and (i=0) then

begin //last page is empty (& not just an empty file) - if we return now we could be causing caller to add another dir page unecessarily

if retryForAccuracy then

begin

{$IFDEF DEBUG_LOG}

log.add(st.who,where+routine,format('Last dir page (%d) is empty, so we retry to wait for initial entry to avoid double adding',[lastDirPage]),vDebugError);

{$ENDIF}

//speed? I suppose we could return with i=1 so the caller tries to use this new page immediately

// seems like a better option but we can't be sure that slot will be free for us!

// - we could just retry this last page loop count (but once we can jump to the last dir page, no speed benefit)

if retry=-1 then

retry:=RETRY_DIRCOUNT;

{$IFDEF DEBUG_LOG}

log.add(st.who,where+routine,format('Retry set to %d',[retry]),vDebugLow);

{$ENDIF}

//exit; //abort - note will cause fail on finally, unpin

continue; //abort/retry

end;

//else don't retry, we have the right count probably -1

end;

prevLastSlotDirPage:=lastDirPage; //all other slots' previous slots are on this page

//since we return the last slot page & the count

// - so it's up the the caller who adds a new dir page to

// set a different prev dir page

result:=ok;

finally

buffer.unpinPage(st,lastDirPage);

end; {try}

end; {retry}

if (result<>ok) and (retry=0) then

{$IFDEF DEBUG_LOG}

log.add(st.who,where+routine,format('Failed after retries',[nil]),vAssertion);

{$ELSE}

;

{$ENDIF}

end; {with}

end; {DirCount}

function TDBFile.DirPage(st:TStmt;DirPageId:PageId;dirSlot:DirSlotId;var pid:PageId;var space:word):integer;

const routine=':DirPage';

page:TPage;

fileDir:TfileDir;

dirPageOffset:integer; //page jump count

dirPage:PageId;

nextDirPage:PageId;

result:=Fail;

if fStartPage=InvalidPageId then

begin

{$IFDEF DEBUG_LOG}

log.add(st.who,where+routine,'Uninitialised start page',vDebugError);

{$ENDIF}

exit;

end;

//SAFETY: check dirSlot>=0!

{Find the appropriate dir page}

dirPageOffset:=dirSlot div FileDirPerBlock; //page offset

nextDirPage:=fStartPage;

with Ttransaction(st.owner).db.owner as TDBserver do

begin

if DirPageId=InvalidPageId then

begin

{Find the appropriate dir page}

while dirPageOffset>0 do

begin

dirPage:=nextDirPage;

if buffer.pinPage(st,dirPage,page)<>ok then

begin

{$IFDEF DEBUG_LOG}

log.add(st.who,where+routine,format('Failed reading dir page %d',[dirPage]),vDebugError);

{$ENDIF}

exit; //abort

end;

try

if page.block.nextPage=InvalidPageId then

begin

{$IFDEF DEBUG_LOG}

log.add(st.who,where+routine,format('Missing next dir page from %d to read dir slot %d',[dirPage,dirSlot]),vAssertion);

{$ENDIF}

exit; //abort

end;

nextDirPage:=page.block.nextPage;

dec(dirPageOffset);

finally

buffer.unpinPage(st,dirPage);

end; {try}

end; {while}

end

else

begin //dir page reference was passed by caller, so we can jump straight to it

nextDirPage:=DirPageId;

end;

{Ok, now get the dir slot}

if buffer.pinPage(st,nextdirPage,page)<>ok then

begin

{$IFDEF DEBUG_LOG}

log.add(st.who,where+routine,format('Failed reading dir page %d',[nextdirPage]),vDebugError);

{$ENDIF}

exit; //abort

end;

try

page.AsBlock(st,(dirSlot mod FileDirPerBlock)*sizeof(fileDir),sizeof(fileDir),@fileDir);

pid:=fileDir.pid; //if = InvalidPageId then read past end - caller to deal with this

space:=fileDir.Space;

{$IFDEF SAFETY}

if (dirSlot<>0) and (pid=InvalidPageId) then //i.e. if initial dirSlot (0) could just be an empty file with no datapages yet

begin

{$IFDEF DEBUG_LOG}

log.add(st.who,where+routine,format('Invalid page (%d) (space=%d) read from dir page %d (slot=%d)',[pid,space,nextdirPage,dirSlot]),vAssertion);

{$ENDIF}

exit; //abort

end;

{$ENDIF}

result:=ok;

finally

buffer.unpinPage(st,nextdirPage);

end; {try}

end; {with}

end; {DirPage}

function TDBFile.DirPageSet(st:TStmt;DirPageId:PageId;dirSlot:DirSlotId;pid:PageId;space:word;AllowOverwrite:boolean):integer;

const routine=':DirPageSet';

page:TPage;

fileDir:TfileDir;

{$IFDEF DEBUG_CHECKDIR}

existingFileDir:TfileDir;

{$ENDIF}

dirPageOffset:integer; //page jump count

dirPage:PageId;

nextDirPage:PageId;

result:=Fail;

if fStartPage=InvalidPageId then

begin

{$IFDEF DEBUG_LOG}

log.add(st.who,where+routine,'Uninitialised start page',vDebugError);

{$ENDIF}

exit;

end;

with Ttransaction(st.owner).db.owner as TDBserver do

begin

if DirPageId=InvalidPageId then

begin

{Find the appropriate dir page}

dirPageOffset:=dirSlot div FileDirPerBlock; //page offset

dirPage:=fStartPage;

nextDirPage:=fStartPage;

while dirPageOffset>0 do

begin

dirPage:=nextDirPage;

if buffer.pinPage(st,dirPage,page)<>ok then

begin

{$IFDEF DEBUG_LOG}

log.add(st.who,where+routine,format('Failed reading dir page %d',[dirPage]),vDebugError);

{$ENDIF}

exit; //abort

end;

try

if page.block.nextPage=InvalidPageId then

begin

{$IFDEF DEBUG_LOG}

log.add(st.who,where+routine,format('Missing next dir page from %d to read dir slot %d',[dirPage,dirSlot]),vDebugError);

{$ENDIF}

exit; //abort

end;

nextDirPage:=page.block.nextPage;

dec(dirPageOffset);

finally

buffer.unpinPage(st,dirPage);

end; {try}

end; {while}

end

else

begin //dir page reference was passed by caller, so we can jump straight to it

nextDirPage:=DirPageId;

end;

{Ok, now set the dir slot}

if buffer.pinPage(st,nextdirPage,page)<>ok then

begin

{$IFDEF DEBUG_LOG}

log.add(st.who,where+routine,format('Failed reading dir page %d',[nextdirPage]),vDebugError);

{$ENDIF}

exit; //abort

end;

try

if pid=InvalidPageId then

begin

{$IFDEF DEBUG_LOG}

log.add(st.who,where+routine,'Cannot add an invalid page id to the page directory',vAssertion);

{$ENDIF}

exit;

end;

fileDir.pid:=pid;

fileDir.Space:=space;

if page.latch(st)=ok then

begin

try

//Keep live to avoid race! {$IFDEF DEBUG_CHECKDIR}

if not AllowOverwrite then

begin

{Check that the page we're setting is still ours, i.e. when adding new ones check no-one else has grabbed our slot}

page.AsBlock(st,(dirSlot mod FileDirPerBlock)*sizeof(existingFileDir),sizeof(existingFileDir),@existingFileDir);

if (existingFileDir.pid<>InvalidPageId) and (existingFileDir.pid<>fileDir.pid) then

begin

{$IFDEF DEBUG_LOG}

log.add(st.who,where+routine,format('Warning: trying to set an existing page id (%d) in the page directory to a new one (%d) - rejecting',[existingFileDir.pid,fileDir.pid]),vDebugError); //i.e. race bug, e.g. during DirPageAdd! = disaster? but we retry...

{$ENDIF}

result:=-2;

exit; //abort! - up to caller to retry

end;

end;

//else assume caller knows what they're doing!

//{$ENDIF}

page.SetBlock(st,(dirSlot mod FileDirPerBlock)*sizeof(fileDir),sizeof(fileDir),@fileDir);

page.dirty:=True;

finally

page.unlatch(st);

end; {try}

//30/08/00 moved:page.dirty:=True;