DIRECTORY
AlpineFS USING [OpenFile, ErrorFromStream, OpenFileFromStream],
BasicTime USING [Now],
FS USING [Error, ErrorDesc, OpenFile,
	ErrorFromStream, GetInfo, SetByteCountAndCreatedTime, SetPageCount],
IO,
RefText USING [TrustTextAsRope],
Rope,
WalnutDefs USING [Error],
WalnutKernelDefs USING [LogEntry],
WalnutRoot USING [CommitAndContinue, GetStreamsForExpunge, ReturnExpungeStreams, StatsReport],
WalnutStream USING [logFileInfo, Aborted, CopyBytes, FindNextEntry, FlushStream, PeekEntry, ReadEntry, SetHighWaterMark, WriteEntry],

WalnutLogExpunge;

WalnutLogExpungeImpl: CEDAR PROGRAM

IMPORTS
AlpineFS, BasicTime, FS, IO, RefText, Rope,
WalnutDefs, WalnutRoot, WalnutStream

EXPORTS
WalnutLogExpunge =

BEGIN


ROPE: TYPE = Rope.ROPE;
STREAM: TYPE = IO.STREAM;
LogEntry: TYPE = WalnutKernelDefs.LogEntry;


currentStream: STREAM;
currentEntryLengthHint: INT;
currentEntryPos: INT;

expungeStream: STREAM;
keyIs: ROPE;
expungeInternalFileID: INT;
logSeqNo: INT;
StartExpunge: PUBLIC PROC[pagesNeeded: INT] RETURNS[expungeFileID: INT] = {
reason: FS.ErrorDesc;
actualPages: INT;
of: FS.OpenFile;

BEGIN ENABLE BEGIN
FS.Error => { reason _ error; GOTO err };
IO.Error => { reason _ AlpineFS.ErrorFromStream[stream]; GOTO err };
UNWIND => { currentStream _ expungeStream _ NIL };
END;

[currentStream, expungeStream, keyIs, expungeInternalFileID, logSeqNo] _
WalnutRoot.GetStreamsForExpunge[];

actualPages _ FS.GetInfo[of _ AlpineFS.OpenFileFromStream[expungeStream]].pages;
IF actualPages < pagesNeeded THEN
FS.SetPageCount[of, pagesNeeded ! FS.Error =>
IF error.code = $quotaExceeded THEN GOTO quota ELSE REJECT];

IF expungeStream.GetLength[] # 0 THEN {
WalnutStream.SetHighWaterMark[expungeStream, 0, -1];
expungeStream.SetIndex[0];
expungeStream.SetLength[0];
};

FS.SetByteCountAndCreatedTime[of, -1, BasicTime.Now[]];
WalnutStream.logFileInfo. key _ keyIs.ToRefText[];
WalnutStream.logFileInfo.internalFileID _ expungeInternalFileID;
WalnutStream.logFileInfo.logSeqNo _ logSeqNo;
[] _ WalnutStream.WriteEntry[expungeStream, WalnutStream.logFileInfo];
WalnutStream.FlushStream[expungeStream, TRUE];

RETURN[expungeInternalFileID];

EXITS
quota => 
ERROR WalnutDefs.Error[$expungeLog, $quotaExceeded,
IO.PutFR[" An ExpungeLog of %g pages exceeds your quota; file has %g pages", IO.int[pagesNeeded], IO.int[actualPages]]];
err => {
WalnutRoot.StatsReport["\n *** During start expunge "];
WalnutRoot.StatsReport[reason.explanation];
ERROR WalnutDefs.Error[$expungeLog, reason.code, "During Start Expunge"];
};
END;
}; 

RestartExpunge: PUBLIC PROC[currentLogPos, expungeLogLength: INT]
RETURNS[ok: BOOL] = {
DoRE: PROC = {
expungePosToUse: INT _ expungeLogLength;
le: LogEntry;
keyFromExpLog: REF TEXT;
internalFileIDFromExpLog, logSeqNoFromExpLog: INT;
reason: FS.ErrorDesc;

BEGIN ENABLE BEGIN
FS.Error => { reason _ error; GOTO error };
IO.Error => { reason _ AlpineFS.ErrorFromStream[stream]; GOTO error };
UNWIND => { currentStream _ expungeStream _ NIL };
END;
ok _ FALSE;

[currentStream, expungeStream, keyIs, expungeInternalFileID, logSeqNo] _
WalnutRoot.GetStreamsForExpunge[];

IF expungeStream = NIL THEN RETURN;
IF expungeStream.GetLength[] < expungeLogLength THEN RETURN;		-- too short

expungeStream.SetIndex[0];
le _ WalnutStream.ReadEntry[expungeStream].le;
IF le = NIL THEN RETURN;
TRUSTED {
WITH le: le SELECT FROM
LogFileInfo => {
keyFromExpLog _ le.key;
internalFileIDFromExpLog _ le.internalFileID;
logSeqNoFromExpLog _ le.logSeqNo;
};
ENDCASE => RETURN;
};
IF expungeLogLength = 0 THEN expungePosToUse _ expungeStream.GetIndex[];
IF ~keyIs.Equal[RefText.TrustTextAsRope[keyFromExpLog]] THEN RETURN;
IF internalFileIDFromExpLog # expungeInternalFileID THEN RETURN;
IF logSeqNoFromExpLog # logSeqNo THEN RETURN;

WalnutStream.SetHighWaterMark[expungeStream, expungePosToUse, -1];

expungeStream.SetIndex[expungePosToUse];
currentStream.SetIndex[currentLogPos];
currentEntryLengthHint _ -1;
currentEntryPos _ currentLogPos;
ok _ TRUE;
EXITS error => {
WalnutRoot.StatsReport["\n *** During Start Expunge "];
WalnutRoot.StatsReport[reason.explanation];
ok _ FALSE;
};
END;
};
DoRE[];
IF ~ok THEN {
WalnutRoot.ReturnExpungeStreams[];
currentStream _ expungeStream _ NIL;
};
};

CopyBytesToExpungeLog: PUBLIC PROC[bytesToCopy: INT] = {
Cbel: PROC = {
expungeStream.SetIndex[expungeStream.GetLength[]];	-- to be sure
WalnutStream.CopyBytes[
from: currentStream, to: expungeStream, num: bytesToCopy];
currentEntryPos _ currentStream.GetIndex[];
currentEntryLengthHint _ -1;		-- unknown
};

CarefullyApply[Cbel, "CopyBytesToExpungeLog"];
};

GetExpungeProgress: PUBLIC PROC RETURNS[currentLogPos, expungeLogLength: INT] = {
Gep: PROC = {
WalnutStream.FlushStream[expungeStream, TRUE];
WalnutRoot.CommitAndContinue[];
currentLogPos _ currentEntryPos;
expungeLogLength _ expungeStream.GetLength[];
};
CarefullyApply[Gep, "GetExpungeProgress"];
};
EndExpunge: PUBLIC PROC =
{ expungeStream _ currentStream _ NIL };

Shutdown: PUBLIC PROC =
{ expungeStream _ currentStream _ NIL };
SetPosition: PUBLIC PROC[startPos: INT] RETURNS[charsSkipped: INT] = {
next: INT;
SetPs: PROC = {
currentStream.SetIndex[startPos];
next _ WalnutStream.FindNextEntry[currentStream];
currentEntryLengthHint _ -1;
IF next # -1 THEN currentEntryPos _ next;
};
CarefullyApply[SetPs, "SetPosition"];
RETURN[IF next = -1 THEN next ELSE next - startPos];
};

SetIndex: PUBLIC PROC[pos: INT] = {
Si: PROC = {
currentStream.SetIndex[pos];
currentEntryLengthHint _ -1;
currentEntryPos _ pos;
};
CarefullyApply[Si, "SetIndex"];
};

PeekEntry: PUBLIC PROC RETURNS[ident: ATOM, msgID: REF TEXT, at: INT] = {
Pne: PROC = {
length: INT;
[ident, msgID, length] _ WalnutStream.PeekEntry[currentStream];
currentEntryLengthHint _ length;
IF length = -1 THEN {
IF currentEntryPos = currentStream.GetLength[]
THEN at _ currentEntryPos ELSE at _ -1;
RETURN;
};
at _ currentEntryPos;
};
CarefullyApply[Pne, "PeekEntry"];
};
SkipEntry: PUBLIC PROC RETURNS[ok: BOOL] = {
Sne: PROC = {
length: INT;
ok _ FALSE;
IF currentEntryLengthHint = -1 THEN
length _ WalnutStream.PeekEntry[currentStream].length
ELSE {
length _ currentEntryLengthHint;
currentEntryLengthHint _ -1;
};
IF length = -1 THEN RETURN;
currentStream.SetIndex[currentEntryPos _ currentStream.GetIndex[] + length];
ok _ TRUE;
};
CarefullyApply[Sne, "SkipEntry"];
};

CopyEntry: PUBLIC PROC RETURNS[newPosition, bytesCopied: INT] = {
Ce: PROC = {
bytesCopied _ 0;
newPosition _ expungeStream.GetIndex[];
IF currentEntryLengthHint = -1 THEN
currentEntryLengthHint _ WalnutStream.PeekEntry[currentStream].length;
IF currentEntryLengthHint = -1 THEN {newPosition _ -1; RETURN};
WalnutStream.CopyBytes[
from: currentStream, to: expungeStream, num: currentEntryLengthHint];

currentEntryPos _ currentStream.GetIndex[];
bytesCopied _ currentEntryLengthHint;
currentEntryLengthHint _ -1;		-- unknown
};
CarefullyApply[Ce, "CopyEntry"];
};

EndCopyEntry: PUBLIC PROC RETURNS[startCopyPos: INT] = {
Ne: PROC[] = {
wle: LogEntry _ WalnutStream.ReadEntry[currentStream].le;
currentStream.SetIndex[currentEntryPos];
IF wle = NIL THEN
ERROR WalnutDefs.Error[$log, $NoEntryFound, "During Expunge"];

TRUSTED { WITH le: wle SELECT FROM
EndCopyNewMailInfo => startCopyPos _ le.startCopyPos;
EndCopyReadArchiveInfo => startCopyPos _ le.startCopyPos;
ENDCASE =>
ERROR WalnutDefs.Error[$log, $WrongEntryFound, "During Expunge"];
};
};
startCopyPos _ 0;
CarefullyApply[Ne, "EndCopyNewMailEntry"];
};

LogLength: PUBLIC PROC RETURNS[length: INT] = {
IF currentStream = NIL THEN RETURN[-1]; 
RETURN[currentStream.GetLength[]];
};

GetIndex: PUBLIC PROC RETURNS[length: INT] = {
IF currentStream = NIL THEN RETURN[-1]; 
RETURN[currentStream.GetIndex[]];
};

CopyBytes: PUBLIC PROC[strm: STREAM, num: INT] = {
Cb: PROC = {
strm.SetIndex[strm.GetLength[]];
WalnutStream.CopyBytes[from: currentStream, to: strm, num: num];
currentEntryPos _ currentStream.GetIndex[];
currentEntryLengthHint _ -1;		-- unknown
};
CarefullyApply[Cb, "CopyBytes"];
};

ExamineThisEntry: PUBLIC PROC RETURNS[status: WalnutLogExpunge.EntryStatus] = {
Ete: PROC = {
length: INT;
curPos: INT;
status _ noValidEntry;
IF currentEntryLengthHint = -1 THEN
length _ WalnutStream.PeekEntry[currentStream].length
ELSE {
length _ currentEntryLengthHint;
currentEntryLengthHint _ -1;
};
IF length = -1 THEN { status _ noValidEntry; RETURN};
status _ validEntry;
currentStream.SetIndex[currentEntryPos _ (curPos _ currentStream.GetIndex[]) + length !
		IO.EndOfStream => { status _ EndOfStream; CONTINUE }];
IF status = EndOfStream THEN currentStream.SetIndex[currentEntryPos _ curPos];
};
CarefullyApply[Ete, "ExamineThisEntry"];
};

CarefullyApply: PROC[proc: PROC[], who: ROPE ] = {
ENABLE BEGIN
IO.Error => IF WalnutStream.Aborted[stream] THEN GOTO aborted ELSE {
ed: FS.ErrorDesc;
ed _ FS.ErrorFromStream[stream ! FS.Error, IO.Error => CONTINUE];
IF ed.code = $quotaExceeded THEN GOTO quota;
WalnutRoot.ReturnExpungeStreams[];
currentStream _ expungeStream _ NIL;
REJECT
};
FS.Error => {
IF error.code = $transAborted THEN GOTO aborted;
IF error.code = $quotaExceeded THEN GOTO quota;
WalnutRoot.ReturnExpungeStreams[];
currentStream _ expungeStream _ NIL;
REJECT
}; 
UNWIND => {
WalnutRoot.ReturnExpungeStreams[];
currentStream _ expungeStream _ NIL;
};
END;

proc[];
EXITS
aborted => {
WalnutRoot.StatsReport["\n **** Trans abort during expunge doing "];
WalnutRoot.StatsReport[who];
WalnutRoot.ReturnExpungeStreams[];
currentStream _ expungeStream _ NIL;
ERROR WalnutDefs.Error[$log, $TransactionAbort, "During expunge"];
};
quota => {
WalnutRoot.StatsReport["\n **** Quota exceeded during expunge doing "];
WalnutRoot.StatsReport[who];
WalnutRoot.ReturnExpungeStreams[];
currentStream _ expungeStream _ NIL;
ERROR WalnutDefs.Error[$log, $QuotaExceeded, "During expunge"];
};
};

END.
¢WalnutLogExpungeImpl.Mesa
Copyright c 1985 by Xerox Corporation.  All rights reserved.
Willie-Sue, October 14, 1985 12:06:22 pm PDT

Contents: types and procedures which take the current and expunge log streams as implicit parameters
Types
Variables
Managing the ExpungeLog
Writes the first log record on the expunge log, from information from the root; returns the internalFileID of the expunge log being written on
Set the length of the expunge log and the read position of the current log to be the given values.  If it returns NOT ok, then you must StartExpunge -- StartExpunge is always guaranteed to succeed (if space exists!)
Copies bytesToCopy from the currentLog at its current position onto the end of the expungeLog;  does not flush the expungeLog (must use GetExpungeProgress)
Get the read position of the current log and the length of the expunge log  (this information can later be used to restart an expunge that was in progress); forces a flush (commit) of the expunge log - raises WalnutDefs.Error with code = $LogTransAbort if the flush fails

Parsing a log for Expunge
returns at = -1 if there is not a valid entry at the current position.  The entry is not "consumed" so that performing a NextEntry or a CopyEntry after reading the ident & msg ID information will copy or retrieve the entry returned.  Performing successive PeekEntry's will NOT advance the log position; use SkipEntry to advance the log
Does not check if ident is for a legal log entry type.

returns NOT ok if there is not a valid entry at the current position, and does not advance the log position.  If there is a valid entry, it is "consumed", and the log is advanced.
Copies the next entry from the current log to the expunge log; it returns the new position that the entry now has in the expunge log;  does NOT flush (commit) the expungeLog, raises WalnutDefs.Error with code = $ExpungeTransAbort if the expunge transaction aborts
returns noValidEntry if there is not a valid entry at the current position, and does not advance the log position.  If there is a valid entry, it is "consumed", and the log is advanced.
IF IO.EndOfStream is raised, EndOfStream is returned and log is not advanced.
Local utilities
�Ê¯˜š
œ™Jšœ
Ïm
œ1™<J™,—J™Jš
œd™dJ˜š
Ïk	˜	Jšœ	žœ1˜?Jšœ
žœ˜šžœ
žœ˜%Jš
œE˜E—Jšžœ
˜Jšœžœ˜ J˜Jšœžœ	˜Jšœžœ˜"JšœžœN˜^Jšœ
žœs˜…
J˜Jš
œ˜—J˜šœžœ
ž˜#J˜š
ž˜Jšœžœžœ˜+Jš
œ$˜$J˜—š
ž˜J˜——J˜š
ž˜J˜š
Ïb™J˜Jšžœžœžœ
˜Jšžœžœžœ
žœ
˜Jšœ
žœ˜+J˜—š
Ÿ	™	J˜Jšœžœ
˜Jšœžœ
˜Jšœžœ
˜J˜Jšœžœ
˜Jšœžœ
˜Jšœžœ
˜Jšœ
žœ
˜——
head1š
Ÿ™š
Ïnœžœžœžœžœ˜KJš
œŽ
™Ž
Jšœžœ˜Jšœ
žœ
˜Jšœžœ
˜J˜šžœ
ž˜Jšžœžœ˜)Jšžœ7žœ˜DJšžœ&žœ˜2Jšžœ
˜J˜š
œH˜HJš
œ"˜"—J˜Jšœžœ@˜Pšžœž˜!šžœ žœ	˜-Jš
žœžœ
žœžœ
žœ˜<——J˜šžœžœ˜'Jš
œ4˜4Jš
œ˜Jš
œ˜J˜—J˜Jšžœ5˜7Jš
œ2˜2Jš
œ@˜@Jš
œ-˜-Jš
œF˜FJšœ(žœ˜.J˜Jšžœ˜J˜š
ž˜š
œ	˜	šžœ.˜3JšžœKžœžœ˜x——š
œ˜J˜7J˜+JšžœD˜IJ˜——Jšžœ
˜—Jš
œ˜—J˜š
 œžœ"žœžœžœ˜WJ™×
š œžœ˜Jšœžœ˜(Jš
œ
˜
Jšœžœ
žœ
˜Jšœ.žœ
˜2Jšœžœ˜J˜šžœ
žœ
ž˜Jšžœžœ	˜+Jšžœ7žœ	˜FJšžœ&žœ˜2Jšžœ
˜—Jšœžœ
˜J˜š
œH˜HJš
œ"˜"—J˜Jšžœžœ
žœ
žœ
˜#Jšžœ.žœÏc
˜JJ˜Jš
œ˜Jš
œ.˜.Jšžœžœ
žœ
žœ
˜šžœ˜	šžœžœ
ž˜š
œ˜Jš
œ˜J˜-J˜!Jš
œ˜—Jšžœžœ
˜—J˜—Jšžœžœ,˜HJšžœ6žœ
žœ
˜DJšžœ2žœ
žœ
˜@Jšžœžœ
žœ
˜-J˜Jš
œB˜BJ˜Jš
œ(˜(Jš
œ&˜&Jš
œ˜Jš
œ ˜ šœžœ
˜
šžœ˜J˜7J˜+Jšœžœ
˜J˜——Jšžœ
˜J˜—J˜šžœžœ˜
J˜"Jšœ žœ
˜$J˜—J˜—J˜š œžœ
žœžœ˜8Jš
œ›
™›
š œžœ˜Jšœ3¡
˜@š
œ˜Jš
œ:˜:—Jš
œ+˜+Jšœ¡
˜(J˜—J˜Jš
œ.˜.J˜—J˜š
 œžœ
žœ
žœ"žœ˜QJš
œ�™�š œžœ˜
Jšœ(žœ˜.J˜Jš
œ ˜ Jš
œ-˜-J˜—Jš
œ*˜*J˜—J™š 
œžœ˜Jšœ"žœ˜(—J˜š œžœ˜Jšœ"žœ˜(——š
Ÿ™š œžœ
žœžœžœžœ˜FJšœžœ
˜
š œžœ˜Jš
œ!˜!Jš
œ1˜1Jš
œ˜Jšžœžœ˜)J˜—Jš
œ%˜%Jšžœ
žœžœžœ˜4J˜—J˜š œžœžœ˜#š œžœ˜Jš
œ˜Jš
œ˜Jš
œ˜Jš
œ˜—Jš
œ˜J˜—J˜š 	œžœ
žœžœ	žœ
žœžœ˜IJš
œ†™†š œžœ˜
Jšœžœ
˜Jš
œ?˜?Jš
œ ˜ šžœ
žœ˜šžœ,˜.Jšžœžœ	˜'Jšžœ
˜J˜——Jš
œ˜J˜—Jš
œ!˜!J˜—J™š 	œžœ
žœžœ˜,Jš
œ³
™³
š œžœ˜
Jšœžœ
˜Jšœžœ
˜šžœž˜#Jš
œ5˜5šžœ˜Jš
œ ˜ Jš
œ˜Jš
œ˜——Jšžœ
žœ
žœ
˜Jš
œL˜LJšœžœ
˜
J˜—Jš
œ!˜!J˜—J˜š 	œžœ
žœžœ˜AJš
œ‡™‡š œžœ˜Jš
œ˜Jš
œ'˜'šžœž˜#Jš
œF˜F—Jšžœžœžœ˜?š
œ˜Jš
œE˜E—J˜Jš
œ+˜+Jš
œ%˜%Jšœ¡
˜(J˜—Jš
œ ˜ Jš
œ˜—J˜š
 œžœ
žœ
žœžœ˜8š œžœ˜Jš
œ9˜9Jš
œ(˜(šžœžœ
ž˜Jšžœ9˜>—J˜šžœžœ	žœ
ž˜"Jš
œ5˜5Jš
œ9˜9šžœ˜
Jšžœ<˜A—J˜—J˜—Jš
œ˜Jš
œ*˜*J˜—J˜š
 	œžœ
žœ
žœ	žœ˜/Jšžœžœ
žœ˜(Jšžœ˜"J˜—J˜š
 œžœ
žœ
žœ	žœ˜.Jšžœžœ
žœ˜(Jšžœ˜!J˜—J˜š
 	œžœ
žœžœžœ˜2š œžœ˜J˜ Jš
œ@˜@Jš
œ+˜+Jšœ¡
˜(J˜—Jš
œ ˜ Jš
œ˜—J˜š œžœ*˜OJš
œ¹
™¹
Jš
œM™Mš œžœ˜
Jšœžœ
˜Jšœžœ
˜Jš
œ˜šžœž˜#Jš
œ5˜5šžœ˜Jš
œ ˜ Jš
œ˜Jš
œ˜——Jšžœ
žœžœ˜5Jš
œ˜š
œW˜WJšœžœ(žœ˜8—Jšžœžœ2˜NJ˜—Jš
œ(˜(J˜—J˜—™š œžœžœ	žœ˜2šžœ
ž˜š
žœ
žœžœ
žœ	žœ˜DJšœžœ˜Jš	œžœžœžœ
žœ˜AJšœžœ
žœ˜,J˜"Jšœ žœ
˜$Jš
ž˜Jš
œ˜—šžœ
ž
˜
Jšžœžœ
žœ	˜0Jšžœžœ
žœ˜/J˜"Jšœ žœ
˜$Jš
ž˜Jš
œ˜—šžœ˜J˜"Jšœ žœ
˜$J˜—Jšžœ
˜J˜J˜š
ž˜š
œ˜J˜DJ˜J˜"Jšœ žœ
˜$Jšžœ=˜BJ˜—š
œ
˜
J˜GJ˜J˜"Jšœ žœ
˜$Jšžœ:˜?J˜———J˜——J˜Jšžœ
˜—…—&,;}