IMPORTS
BasicTime, FS, IO, MailUtils, RefText, Rope, RuntimeError, SendMailOps,
WalnutDefs, WalnutRoot, WalnutStream
Types
GMT: TYPE = BasicTime.GMT;
ROPE: TYPE = Rope.ROPE;
STREAM: TYPE = IO.STREAM;
LogEntry: TYPE = WalnutKernelDefs.LogEntry;
MsgLogEntry: TYPE = WalnutKernelDefs.MsgLogEntry;
TiogaContents: TYPE = ViewerTools.TiogaContents;
LogInfo: TYPE = WalnutDefs.LogInfo;
WalnutOpsHandle: TYPE = WalnutDefs.WalnutOpsHandle;
LogHandle: TYPE = WalnutLog.LogHandle;
LogHandleRec: PUBLIC TYPE = WalnutLog.LogHandleRec;
RootHandle: TYPE = WalnutRoot.RootHandle;
RootHandleRec: PUBLIC TYPE = WalnutRoot.RootHandleRec;
InternalLogInfo: TYPE = WalnutRoot.InternalLogInfo;
the all streams are under the same transaction; WalnutLogImpl must get them WalnutRoot to open the files
Logging of operations that perform actions on the Walnut database
ExpungeMsgs:
PUBLIC
PROC[opsH: WalnutOpsHandle]
RETURNS[at, next:
INT] =
{ [at, next] ¬ WriteEntry[opsH, WalnutStream.logInfoRef.expungeMsgs] };
WriteExpungeLog:
PUBLIC
PROC[opsH: WalnutOpsHandle]
RETURNS[at, next:
INT] = {
[at, next] ¬ WriteEntry[opsH, WalnutStream.logInfoRef.writeExpungeLog];
};
CreateMsgSet:
PUBLIC
PROC[opsH: WalnutOpsHandle, name:
ROPE]
RETURNS[at, next:
INT] = {
WalnutStream.logInfoRef.createMsgSet.msgSet ¬ name;
[at, next] ¬ WriteEntry[opsH, WalnutStream.logInfoRef.createMsgSet]
};
EmptyMsgSet:
PUBLIC
PROC[opsH: WalnutOpsHandle, msgSet:
ROPE]
RETURNS[at, next: INT] = {
WalnutStream.logInfoRef.emptyMsgSet.msgSet ¬ msgSet;
[at, next] ¬ WriteEntry[opsH, WalnutStream.logInfoRef.emptyMsgSet]
};
DestroyMsgSet:
PUBLIC
PROC[opsH: WalnutOpsHandle, msgSet:
ROPE]
RETURNS[at, next:
INT] = {
WalnutStream.logInfoRef.destroyMsgSet.msgSet ¬ msgSet;
[at, next] ¬ WriteEntry[opsH, WalnutStream.logInfoRef.destroyMsgSet]
};
AddMsg:
PUBLIC
PROC[opsH: WalnutOpsHandle, msg, to:
ROPE]
RETURNS[at, next:
INT] = {
WalnutStream.logInfoRef.addMsg.msg ¬ msg;
WalnutStream.logInfoRef.addMsg.to ¬ to;
[at, next] ¬ WriteEntry[opsH, WalnutStream.logInfoRef.addMsg]
};
RemoveMsg:
PUBLIC
PROC[opsH: WalnutOpsHandle, msg, from:
ROPE]
RETURNS[at, next: INT] = {
WalnutStream.logInfoRef.removeMsg.msg ¬ msg;
WalnutStream.logInfoRef.removeMsg.from ¬ from;
[at, next] ¬ WriteEntry[opsH, WalnutStream.logInfoRef.removeMsg]
};
MoveMsg:
PUBLIC
PROC[opsH: WalnutOpsHandle, msg, from, to:
ROPE]
RETURNS[at, next: INT] = {
WalnutStream.logInfoRef.moveMsg.msg ¬ msg;
WalnutStream.logInfoRef.moveMsg.from ¬ from;
WalnutStream.logInfoRef.moveMsg.to ¬ to;
[at, next] ¬ WriteEntry[opsH, WalnutStream.logInfoRef.moveMsg]
};
HasBeenRead:
PUBLIC
PROC[opsH: WalnutOpsHandle, msg:
ROPE]
RETURNS[at, next:
INT] = {
WalnutStream.logInfoRef.hasbeenRead.msg ¬ msg;
[at, next] ¬ WriteEntry[opsH, WalnutStream.logInfoRef.hasbeenRead]
};
RecordNewMailInfo:
PUBLIC
PROC[opsH: WalnutOpsHandle, logLen:
INT, when:
GMT, server:
ROPE, num:
INT]
RETURNS[at, next: INT] = {
WalnutStream.logInfoRef.recordNewMailInfo.logLen ¬ logLen;
WalnutStream.logInfoRef.recordNewMailInfo.when ¬ when;
WalnutStream.logInfoRef.recordNewMailInfo.server ¬ server;
WalnutStream.logInfoRef.recordNewMailInfo.num ¬ num;
[at, next] ¬ WriteEntry[opsH, WalnutStream.logInfoRef.recordNewMailInfo]
};
StartCopyNewMail:
PUBLIC
PROC[opsH: WalnutOpsHandle]
RETURNS[at, next:
INT] =
{ [at, next] ¬ WriteEntry[opsH, WalnutStream.logInfoRef.startCopyNewMail] };
AcceptNewMail:
PUBLIC
PROC[opsH: WalnutOpsHandle]
RETURNS[at, next:
INT] =
{ [at, next] ¬ WriteEntry[opsH, WalnutStream.logInfoRef.acceptNewMail] };
StartReadArchiveFile:
PUBLIC
PROC[opsH: WalnutOpsHandle, file:
ROPE, msgSet:
ROPE]
RETURNS[at, next:
INT] = {
WalnutStream.logInfoRef.startReadArchiveFile.file ¬ file;
WalnutStream.logInfoRef.startReadArchiveFile.msgSet ¬ msgSet;
[at, next] ¬ WriteEntry[opsH, WalnutStream.logInfoRef.startReadArchiveFile]
};
EndReadArchiveFile:
PUBLIC
PROC[opsH: WalnutOpsHandle]
RETURNS[at, next:
INT] =
{ [at, next] ¬ WriteEntry[opsH, WalnutStream.logInfoRef.endReadArchiveFile] };
StartCopyReadArchive:
PUBLIC
PROC[opsH: WalnutOpsHandle]
RETURNS[at, next:
INT] =
{ [at, next] ¬ WriteEntry[opsH, WalnutStream.logInfoRef.startCopyReadArchive] };
Managing the current log
InitLogStream:
PUBLIC
PROC[opsH: WalnutOpsHandle]
RETURNS[
BOOL] = {
opsH.logHandle ¬ NEW[LogHandleRec ¬ [] ];
OpenLogStreams[opsH];
RETURN[TRUE];
};
LogLength:
PUBLIC
PROC[opsH: WalnutOpsHandle]
RETURNS[length:
INT] = {
li: InternalLogInfo = opsH.rootHandle.currentLog;
RETURN[ IF li.writeStream # NIL THEN li.writeStream.GetLength[] ELSE li.readStream.GetLength[] ];
};
OpenLogStreams:
PUBLIC
PROC[opsH: WalnutOpsHandle] = {
IF opsH.rootHandle.currentLog =
NIL
OR opsH.rootHandle.currentLog.readStream =
NIL
THEN {
WalnutRoot.OpenLogStreams[opsH];
IF opsH.logHandle = NIL THEN opsH.logHandle ¬ NEW[LogHandleRec ¬ [] ];
IF opsH.logHandle.scanning THEN SetLogIndex[opsH, opsH.logHandle.entryPos, TRUE];
};
};
ReleaseWriteLock:
PUBLIC
PROC[opsH: WalnutOpsHandle] = {
logH: LogHandle = opsH.logHandle;
IF opsH.rootHandle.currentLog.writeStream = NIL THEN RETURN;
WalnutRoot.ReleaseWriteLock[opsH];
IF logH.scanning THEN SetLogIndex[opsH, logH.entryPos, TRUE];
};
ResetLog:
PUBLIC
PROC[opsH: WalnutOpsHandle, newLength:
INT] = {
strm: STREAM;
IF opsH.logHandle = NIL THEN RETURN;
SetLogIndex[opsH, newLength, TRUE];
WalnutStream.SetHighWaterMark[strm ¬ opsH.rootHandle.currentLog.writeStream,
newLength, -1];
strm.SetLength[newLength];
strm.Flush[];
April 26, 1991, DCS, assure invariant that log streams are positioned at end, except temporarily when explicitly repositioned.
strm.SetIndex[newLength];
opsH.logHandle.scanning ¬ FALSE;
};
ForgetLogStreams:
PUBLIC
PROC[opsH: WalnutOpsHandle] = {
logH: LogHandle = opsH.logHandle;
IF logH = NIL THEN RETURN;
logH.scanning ¬ FALSE;
logH.strm ¬ NIL;
logH.scanningLog ¬ NIL;
};
ShutdownLog:
PUBLIC
PROC[opsH: WalnutOpsHandle] = {
logH: LogHandle = opsH.logHandle;
IF logH = NIL THEN RETURN;
logH.scanning ¬ FALSE;
logH.strm ¬ NIL;
logH.scanningLog ¬ NIL;
WalnutRoot.CloseTransaction[opsH];
};
Write a message on the current log
WriteMessage:
PUBLIC
PROC[opsH: WalnutOpsHandle, msg:
ROPE, body: TiogaContents]
RETURNS[at: INT]= {
wStrm: STREAM = opsH.rootHandle.currentLog.writeStream;
wStrm.SetIndex[wStrm.GetLength[]];
WalnutStream.logInfoRef.createMsg.msg ¬ msg;
IF ( body.formatting.Length[] # 0 )
AND ( body.contents.Fetch[0] = '\r
OR body.contents.Fetch[0] = '\l )
THEN {
last: INT ¬ body.contents.Length[] - 1;
IF body.contents.Fetch[last] = '\000
THEN
-- NUL for padding
{ body.contents ¬ Rope.Substr[body.contents, 1, last-1];
body.formatting ¬ Rope.Concat["\000", body.formatting]
}
ELSE
IF body.contents.Fetch[0] = '\r
OR body.contents.Fetch[0] = '\l
THEN
body.contents ¬ Rope.Substr[body.contents, 1]
};
WalnutStream.logInfoRef.createMsg.textLen ¬ body.contents.Length[];
WalnutStream.logInfoRef.createMsg.formatLen ¬ body.formatting.Length[];
at ¬ WalnutStream.WriteEntry[wStrm, WalnutStream.logInfoRef.createMsg];
WalnutStream.WriteMsgBody[wStrm, body];
WalnutStream.FlushStream[strm: wStrm, setCreateDate: TRUE];
};
Parsing a log (used by Archive and Replay/Scavenge
SetPosition:
PUBLIC
PROC[opsH: WalnutOpsHandle, startPos:
INT]
RETURNS[charsSkipped: INT] = {
next: INT;
logH: LogHandle = opsH.logHandle;
SetLogIndex[opsH, startPos, TRUE];
next ¬ WalnutStream.FindNextEntry[logH.strm];
logH.entryLengthHint ¬ -1;
IF next # -1 THEN logH.entryPos ¬ next;
RETURN[IF next = -1 THEN next ELSE next - startPos];
};
SetIndex:
PUBLIC PROC[opsH: WalnutOpsHandle, pos:
INT] = {
logH: LogHandle = opsH.logHandle;
SetLogIndex[opsH, pos, TRUE];
logH.entryLengthHint ¬ -1;
logH.entryPos ¬ pos;
};
NextAt:
PUBLIC
PROC[opsH: WalnutOpsHandle]
RETURNS[at:
INT] =
{ RETURN[opsH.logHandle.entryPos] };
NextEntry:
PUBLIC
PROC[opsH: WalnutOpsHandle]
RETURNS[le: LogEntry, at:
INT] =
{ [le, at] ¬ NextEntryInternal[opsH, FALSE] };
NextEntryInternal:
PROC[opsH: WalnutOpsHandle, quick:
BOOL]
RETURNS[le: LogEntry, at: INT] = {
logH: LogHandle = opsH.logHandle;
length, startPos: INT;
IF ~logH.scanning THEN ERROR WalnutDefs.Error[$log, $NotScanning];
[le, length] ¬ WalnutStream.ReadEntry[logH.strm, quick];
startPos ¬ logH.entryPos;
logH.entryLengthHint ¬ -1;
IF length # -1 THEN logH.entryPos ¬ startPos + length; -- normal case
RETURN[le, startPos];
};
QuickScan:
PUBLIC
PROC[opsH: WalnutOpsHandle]
RETURNS[le: LogEntry, at:
INT] =
{ [le, at] ¬ NextEntryInternal[opsH: opsH, quick: TRUE] };
ArchiveEntry:
PUBLIC
PROC[opsH: WalnutOpsHandle, to:
STREAM]
RETURNS[ok:
BOOL]= {
Copies the next entry of the current log to an archive stream.
logH: LogHandle = opsH.logHandle;
IF ~logH.scanning THEN ERROR WalnutDefs.Error[$log, $NotScanning];
ok ¬ FALSE;
IF logH.entryLengthHint = -1
THEN
logH.entryLengthHint ¬
WalnutStream.PeekEntry[logH.strm, TRUE].length;
IF logH.entryLengthHint = -1 THEN RETURN;
WalnutStream.CopyBytes[
from: logH.strm, to: to, num: logH.entryLengthHint ];
logH.entryPos ¬ logH.strm.GetIndex[];
logH.entryLengthHint ¬ -1; -- unknown
ok ¬ TRUE;
};
CopyBytesToArchive:
PUBLIC
PROC[
opsH: WalnutOpsHandle, to: STREAM, startPos, num: INT] = {
SetLogIndex[opsH, startPos, FALSE];
WalnutStream.CopyBytes[from: opsH.logHandle.strm, to: to, num: num];
};
Copying NewMail/ReadArchive logs
PrepareToCopyTempLog:
PUBLIC
PROC[opsH: WalnutOpsHandle, which: WalnutKernelDefs.WhichTempLog, pagesAlreadyCopied:
INT, reportProc: WalnutDefs.CheckReportProc]
RETURNS[
BOOL] = {
makes sure the writeStream is long enought to copy the which log onto
RETURN[WalnutRoot.PrepareToCopyTempLog[opsH, which, pagesAlreadyCopied, reportProc] ];
};
CopyTempLog:
PUBLIC
PROC[opsH: WalnutOpsHandle, which: WalnutKernelDefs.WhichTempLog, startCopyPos, fromPos:
INT, reportProc: WalnutDefs.CheckReportProc ] = {
copies the which templog, starting at fromPos, to the end of the writeStream; writes an EndCopy log entry, and sets the length of which to 0; if the operation fails, WalnutDefs.Error is raised; NOT done inside carefullyApply, so that it can catch transaction aborts
exp: ROPE;
endLE: LogEntry;
abort: BOOL ¬ TRUE;
ok: BOOL ¬ FALSE;
copyStrm, wStrm: STREAM;
bytesToCopy: INT;
bytesPerCopy: INT = FS.BytesForPages[200];
opsH.logHandle.scanning ¬ FALSE;
BEGIN
ENABLE
BEGIN
IO.Error => {
ed: FS.ErrorDesc;
<<ed ¬ FS.ErrorFromStream[stream ! IO.Error => CONTINUE];>>
IF ed.code = $transAborted
OR ed.code = $remoteCallFailed
THEN
{ exp ¬ ed.explanation; GOTO exit };
REJECT;
};
FS.Error =>
{
IF error.code = $transAborted THEN GOTO exit;
IF error.code = $remoteCallFailed
THEN {
exp ¬ error.explanation;
GOTO exit;
};
REJECT;
};
END;
SELECT which
FROM
newMail => {
WalnutStream.logInfoRef.endCopyNewMailInfo.startCopyPos ¬ startCopyPos;
endLE ¬ WalnutStream.logInfoRef.endCopyNewMailInfo;
};
readArchive => {
WalnutStream.logInfoRef.endCopyReadArchiveInfo.startCopyPos ¬ startCopyPos;
endLE ¬ WalnutStream.logInfoRef.endCopyReadArchiveInfo
};
ENDCASE => ERROR WalnutDefs.Error[$log, $UnknownLogFileType];
copyStrm ¬ WalnutRoot.GetStreamsForCopy[opsH, which];
wStrm ¬ opsH.rootHandle.currentLog.writeStream;
Play it safe! DCS April 26, 1991
wStrm.SetIndex[wStrm.GetLength[]];
bytesToCopy ¬ copyStrm.GetLength[] - fromPos;
IF bytesToCopy > 0
THEN {
first: BOOL ¬ TRUE;
copyStrm.SetIndex[fromPos];
<<wStrm.SetIndex[wStrm.GetLength[]];>> -- Moved outside conditional.
DO
thisCopy: INT ¬ MIN[bytesPerCopy, bytesToCopy];
WalnutStream.CopyBytes[to: wStrm, from: copyStrm, num: thisCopy];
WalnutStream.FlushStream[wStrm, TRUE];
WalnutRoot.CommitAndContinue[opsH];
IF first THEN first ¬ FALSE ELSE reportProc[opsH, "#"];
bytesToCopy ¬ bytesToCopy - thisCopy;
IF bytesToCopy = 0 THEN EXIT;
ENDLOOP;
};
[] ¬ WalnutStream.WriteEntry[wStrm, endLE];
WalnutStream.FlushStream[wStrm, TRUE];
WalnutRoot.CommitAndContinue[opsH];
EXITS
exit => {
who: ROPE ¬ IF which = newMail THEN "NewMail" ELSE "ReadArchive";
IF ~abort THEN ERROR WalnutDefs.Error[$log, $AccessFailed, who];
WalnutRoot.AbortTempCopy[opsH, which];
WalnutRoot.CloseTransaction[opsH];
ERROR WalnutDefs.Error[
$log, IF exp = NIL THEN $TransactionAbort ELSE $RemoteCallFailed,
IO.PutFR1["Copying the %g log", [rope[who]]] ];
};
END;
WalnutRoot.FinishCopy[opsH, which];
};
FinishTempLogCopy:
PUBLIC
PROC[opsH: WalnutOpsHandle, which: WalnutKernelDefs.WhichTempLog] =
{ WalnutRoot.FinishCopy[opsH, which] };
CreateArchiveLog:
PUBLIC
PROC[
opsH: WalnutOpsHandle, fileToRead:
STREAM, msgSet:
ROPE, reportProc: WalnutDefs.CheckReportProc]
RETURNS[ok:
BOOL] = {
reason: ROPE;
[ok, reason] ¬ CreateReadArchiveLog[opsH, fileToRead, msgSet, reportProc];
IF ok THEN [] ¬ WriteEntry[opsH, WalnutStream.logInfoRef.endReadArchiveFile];
WalnutRoot.ReturnReadArchiveStream[opsH];
};
Utilities
GetTiogaContents:
PUBLIC
PROC[opsH: WalnutOpsHandle, textStart, textLen, formatLen:
INT]
RETURNS[contents: TiogaContents] = {
logH: LogHandle = opsH.logHandle;
IF formatLen # 0
THEN
-- tioga formatting nonsense
{ textLen ¬ textLen + 1; textStart ¬ textStart - 1};
SetLogIndex[opsH, textStart, FALSE];
contents ¬ NEW[ViewerTools.TiogaContentsRec];
contents.contents ¬ WalnutStream.ReadRope[logH.strm, textLen];
contents.formatting ¬ WalnutStream.ReadRope[logH.strm, formatLen];
};
GetRefTextFromLog:
PUBLIC
PROC[opsH: WalnutOpsHandle, startPos, length:
INT, text:
REF
TEXT] = {
natLen: NAT;
bytesRead: NAT;
logH: LogHandle = opsH.logHandle;
CheckSize: PROC[len: INT] RETURNS[nat: NAT] = { nat ¬ len };
natLen ¬ CheckSize[length ! RuntimeError.BoundsFault => GOTO noGood];
SetLogIndex[opsH, startPos, FALSE];
bytesRead ¬ logH.strm.GetBlock[block: text, startIndex: 0, count: natLen];
IF bytesRead < natLen THEN text.length ¬ 0;
EXITS
noGood =>
ERROR WalnutDefs.Error[$log, $BadLength,
IO.PutFR1[" Can't convert %g into a nat", [integer[length]] ]];
};
Writing and Reading NewMail/ReadArchive files
GetNewMailLog:
PUBLIC
PROC[opsH: WalnutOpsHandle, lengthRequired, pagesWanted:
INT]
RETURNS[STREAM] = {
opens the NewMailLog (aborting any current stream that might be open); truncates it to the required length; returns NIL if the log is shorter than required, or if the file could not be opened
BEGIN
ENABLE
IO.Error,
FS.Error => {
AbortMailLog[opsH];
GOTO exit;
};
AbortMailLog[opsH];
IF NOT WalnutRoot.GetNewMailStream[opsH, lengthRequired, pagesWanted].ok THEN RETURN[NIL];
EXITS
exit => AbortMailLog[opsH];
END;
RETURN[opsH.rootHandle.newMailLog.stream];
};
CloseNewMailLog:
PUBLIC
PROC[opsH: WalnutOpsHandle] = {
mStrm: STREAM = opsH.rootHandle.newMailLog.stream;
IF mStrm = NIL THEN RETURN;
WalnutRoot.ReturnNewMailStream[opsH];
};
AbortMailLog:
PROC[opsH: WalnutOpsHandle] = {
mStrm: STREAM = opsH.rootHandle.newMailLog.stream;
IF mStrm = NIL THEN RETURN;
WalnutStream.AbortStream[mStrm ! IO.Error, FS.Error => CONTINUE];
WalnutRoot.ReturnNewMailStream[opsH];
};
CreateReadArchiveLog:
PUBLIC
PROC[opsH: WalnutOpsHandle, fileToRead:
STREAM, msgSet:
ROPE, reportProc: WalnutDefs.CheckReportProc]
RETURNS[ok: BOOL, reason: ROPE] = {
reads file and writes the ReadArchiveLogFile; returns ok if all went well
readArchiveStrm: STREAM;
lastCommitPosInFileToRead: INT;
fileToReadPages: INT ¬ FS.PagesForBytes[fileToRead.GetLength[]];
possibleArchiveLength: INT ¬ MAX[20, fileToReadPages + (fileToReadPages/5)]; -- 120%
ok ¬ FALSE;
reason ¬ NIL;
BEGIN
ENABLE
BEGIN
IO.Error => {
<<reason ¬ FS.ErrorFromStream[fileToRead].explanation;>>
<<IF reason = NIL THEN reason ¬ FS.ErrorFromStream[readArchiveStrm].explanation;>>
IF reason = NIL THEN reason ¬ "IO error during createReadArchiveLog";
GOTO exit;
};
FS.Error => { reason ¬ error.explanation; GOTO exit };
UNWIND => GOTO exit;
END;
IF (readArchiveStrm ¬ WalnutRoot.GetReadArchiveStream[opsH, possibleArchiveLength]) =
NIL
THEN
RETURN[FALSE, "Can't get readArchiveStream"];
readArchiveStrm.SetIndex[0];
WalnutStream.SetHighWaterMark[strm: readArchiveStrm, hwmBytes: 0, numPages: -1];
WalnutRoot.CommitAndContinue[opsH];
[ok, lastCommitPosInFileToRead] ¬ ArchiveReader[
opsH: opsH, archiveStream: readArchiveStrm,
fileToRead: fileToRead,
msgSet: msgSet, reportProc: reportProc, posToStartInFileToRead: 0];
IF ~ok THEN WalnutRoot.ReturnReadArchiveStream[opsH];
EXITS
exit =>
IF readArchiveStrm #
NIL
THEN {
WalnutRoot.ReturnReadArchiveStream[opsH];
};
END;
};
CloseReadArchiveLog:
PUBLIC
PROC[opsH: WalnutOpsHandle] = {
WalnutRoot.ReturnReadArchiveStream[opsH];
};