-- Transport Mechanism Registration Server - Internal Mail.
-- [Indigo]<Grapevine>MS>RegMail.mesa
-- Randy Gobbel, 19-May-81 12:26:40
-- Andrew Birrell, 18-Nov-82 10:33:12
-- Mike Schroeder 8-Feb-83 9:43:22
DIRECTORY
BodyDefs USING[ ItemHeader, ItemLength, ItemType, maxRNameLength,
oldestTime, Password, RName, RNameSize, Timestamp],
EnquiryDefs USING[ ],
HeapDefs USING[ HeapEndWrite, HeapEndRead, HeapReadData,
HeapStartWrite, HeapWriteData, ObjectNumber,
ReaderHandle, WriterHandle],
LocalNameDefs USING[ ReadRSName ],
LogDefs USING[ WriteLogEntry ],
ObjectDirDefs USING[ FreeObject, noObject ],
PolicyDefs USING[ CheckOperation, EndOperation, WaitOperation ],
Process USING[ Detach, DisableTimeout ],
ProtocolDefs USING[ AppendTimestamp, ReturnCode, RSOperation ],
RegBTreeDefs USING[ Lookup, LookupReason, RegistryObject ],
RegServerDefs USING[ IsMember, ReadRegistryMembers, Update,
UpdateSublist ],
RegistryDefs USING[ EnumerateRList],
RetrieveDefs USING[ AccessProcs, Failed, Handle, Create, MailboxState,
NewUser, NextServer, ServerName, ServerState,
WaitForMail ],
SendDefs USING[ Abort, AddRecipient, AddToItem, Create, Destroy,
Handle, StartItem, StartSend, Send ],
String USING[ AppendString, EquivalentString],
VMDefs USING[ AllocatePage, Page, pageSize, Release];
RegMail: MONITOR
IMPORTS BodyDefs, HeapDefs, LocalNameDefs, LogDefs, ObjectDirDefs, PolicyDefs,
Process, RegBTreeDefs, RegistryDefs, RegServerDefs,
RetrieveDefs, SendDefs, String, VMDefs
EXPORTS EnquiryDefs--LoginRSMail--, RegServerDefs =
BEGIN
OPEN ProtocolDefs;
-- the update mail is generated in a separate process to allow return to
-- client sooner, and to avoid deadlock if we generate an update while
-- determining our R-Server name.
updateEntry: BodyDefs.RName = [BodyDefs.maxRNameLength];
updateStamp: BodyDefs.Timestamp;
updateElement: BodyDefs.RName = [BodyDefs.maxRNameLength];
updateOp: ProtocolDefs.RSOperation;
updateRSMailObj: HeapDefs.ObjectNumber;
updatePending: BOOLEAN ← FALSE;
updateCond: CONDITION;
updateEnabled: BOOLEAN ← FALSE; -- set TRUE early in restart --
NoRegistryForName: ERROR = CODE;
FailedToSendUpdate: ERROR = CODE;
EnableUpdate: ENTRY PROC = INLINE
{ updateEnabled ← TRUE };
MailUpdate: PUBLIC ENTRY PROCEDURE [entry: BodyDefs.RName,
stamp: BodyDefs.Timestamp,
element: BodyDefs.RName,
op: ProtocolDefs.RSOperation,
rsMailObj: HeapDefs.ObjectNumber] =
BEGIN
IF updateEnabled
THEN BEGIN
WHILE updatePending DO WAIT updateCond ENDLOOP;
updateEntry.length ← 0;
String.AppendString[updateEntry, entry];
updateStamp ← stamp;
updateElement.length ← 0;
IF element # NIL
THEN String.AppendString[updateElement, element];
updateOp ← op;
updateRSMailObj ← rsMailObj;
updatePending ← TRUE;
NOTIFY updateCond;
END
ELSE IF rsMailObj # ObjectDirDefs.noObject
THEN ObjectDirDefs.FreeObject[rsMailObj];
END;
SendRSMail: ENTRY PROC =
BEGIN
DO UNTIL updatePending DO WAIT updateCond ENDLOOP;
SendUpdate[updateEntry, updateStamp, updateElement, updateOp];
IF updateRSMailObj # ObjectDirDefs.noObject
THEN ObjectDirDefs.FreeObject[updateRSMailObj];
updatePending ← FALSE;
NOTIFY updateCond;
ENDLOOP;
END;
briefUpdate: BodyDefs.ItemType = LOOPHOLE[2001B];
briefUpdateAllowed: BOOLEAN ← TRUE;
SendUpdate: INTERNAL PROC[entry: BodyDefs.RName, stamp: BodyDefs.Timestamp,
element: BodyDefs.RName, op: ProtocolDefs.RSOperation] =
BEGIN
-- mail the object to other registration servers for entry's registry.
regobj: RegBTreeDefs.RegistryObject =
RegBTreeDefs.Lookup[entry, readAny];
myName: BodyDefs.RName;
myPassword: STRING;
[myName, myPassword,] ← LocalNameDefs.ReadRSName[];
IF regobj.type # notFound
THEN BEGIN
message: SendDefs.Handle = SendDefs.Create[];
itsRegReader: HeapDefs.ReaderHandle = LookupRegistry[entry];
IF SendDefs.StartSend[handle: message, senderPwd: myPassword,
sender: myName, returnTo: "DeadLetter.MS"L,
validate: FALSE] # ok
THEN ERROR FailedToSendUpdate[];
IF CopyRList[message, myName, itsRegReader] > 0
THEN BEGIN
humanHint: STRING = "RS internal mail for R-Name "L;
SELECT op FROM
AddMember, DeleteMember =>
IF briefUpdateAllowed
THEN BriefUpdate[message, entry, stamp, element, op]
ELSE FullUpdate[message, regobj.reader];
ENDCASE => FullUpdate[message, regobj.reader];
HeapDefs.HeapEndRead[regobj.reader];
-- in case it gets to DeadLetter! --
SendDefs.StartItem[message, Text];
SendDefs.AddToItem[message, DESCRIPTOR[ @(humanHint.text),
humanHint.length ] ];
SendDefs.AddToItem[message, DESCRIPTOR[ @(entry.text),
entry.length]];
SendDefs.Send[message];
END
ELSE BEGIN --no recipients--
SendDefs.Abort[message]; HeapDefs.HeapEndRead[regobj.reader];
END;
SendDefs.Destroy[message];
END;
END;
LookupRegistry: PROCEDURE[name: BodyDefs.RName]
RETURNS[ reader: HeapDefs.ReaderHandle ] =
BEGIN
oldTimePtr: BodyDefs.Timestamp ← BodyDefs.oldestTime; --ugh!--
rc: ProtocolDefs.ReturnCode;
[reader,rc] ← RegServerDefs.ReadRegistryMembers[name, @oldTimePtr];
IF rc.code # done THEN ERROR NoRegistryForName[];
END;
CopyRList: PROCEDURE[ message: SendDefs.Handle, me: BodyDefs.RName,
reader: HeapDefs.ReaderHandle]
RETURNS[ recipients: CARDINAL ] =
BEGIN
Work: PROC[name: BodyDefs.RName] RETURNS[done:BOOLEAN] =
BEGIN
done ← FALSE;
IF NOT String.EquivalentString[me, name]
THEN BEGIN
SendDefs.AddRecipient[message, name];
recipients ← recipients + 1;
END;
END;
recipients ← 0;
RegistryDefs.EnumerateRList[reader, Work];
HeapDefs.HeapEndRead[reader];
END;
FullUpdate: PROCEDURE[message: SendDefs.Handle, reader: HeapDefs.ReaderHandle] =
BEGIN
buffer: VMDefs.Page = VMDefs.AllocatePage[];
ended: BOOLEAN;
used: CARDINAL;
SendDefs.StartItem[message, updateItem ];
DO [ended,used] ← HeapDefs.HeapReadData[reader, [buffer, VMDefs.pageSize]];
SendDefs.AddToItem[message,
DESCRIPTOR[LOOPHOLE[buffer,POINTER], 2*used] ];
IF ended THEN EXIT;
ENDLOOP;
VMDefs.Release[buffer];
END;
BriefUpdate: PROCEDURE[message: SendDefs.Handle, entry: BodyDefs.RName,
stamp: BodyDefs.Timestamp,
element: BodyDefs.RName, op: ProtocolDefs.RSOperation] =
BEGIN
SendDefs.StartItem[message, briefUpdate ];
SendDefs.AddToItem[message,
DESCRIPTOR[LOOPHOLE[@op,POINTER], 2*SIZE[ProtocolDefs.RSOperation]]];
SendDefs.AddToItem[message,
DESCRIPTOR[LOOPHOLE[entry,POINTER], 2*BodyDefs.RNameSize[entry]]];
SendDefs.AddToItem[message,
DESCRIPTOR[LOOPHOLE[element,POINTER], 2*BodyDefs.RNameSize[element]]];
SendDefs.AddToItem[message,
DESCRIPTOR[LOOPHOLE[@stamp,POINTER], 2*SIZE[BodyDefs.Timestamp]]];
END;
-- RS Mail reading --
handle: RetrieveDefs.Handle = RetrieveDefs.Create[pollingInterval: 30];
readerActive: BOOLEAN ← FALSE;
readerInactive: CONDITION ← [timeout:0];
LoginRSMail: PUBLIC ENTRY PROC =
BEGIN
myName: BodyDefs.RName;
myPassword: STRING;
WHILE readerActive DO WAIT readerInactive ENDLOOP;
[myName, myPassword,] ← LocalNameDefs.ReadRSName[];
RetrieveDefs.NewUser[handle: handle, user: myName,
password: myPassword];
DO sName: BodyDefs.RName = [BodyDefs.maxRNameLength];
noMore: BOOLEAN;
[noMore,,] ← RetrieveDefs.NextServer[handle];
IF noMore THEN EXIT;
LogInboxSite["RS inbox-site: "L];
ENDLOOP;
END;
LogInboxSite: INTERNAL PROC[s: STRING] =
BEGIN
log: STRING = [64];
server: STRING = [64];
RetrieveDefs.ServerName[handle, server];
String.AppendString[log, s];
server.length ← MIN [ server.length, log.maxlength - log.length];
String.AppendString[log, server];
LogDefs.WriteLogEntry[log];
END;
LockReader: ENTRY PROC = INLINE
{ readerActive ← TRUE };
UnlockReader: ENTRY PROC = INLINE
{ readerActive ← FALSE; NOTIFY readerInactive };
MyNameIsBad: ERROR = CODE;
MyPwdIsBad: ERROR = CODE;
ReadMail: PUBLIC PROCEDURE =
BEGIN
-- Reads outstanding mail before returning --
LoginRSMail[];
SELECT RetrieveDefs.MailboxState[handle] FROM
badName => ERROR MyNameIsBad[];
badPwd => ERROR MyPwdIsBad[];
notEmpty =>
IF PolicyDefs.CheckOperation[RSReadMail]
THEN { RegMailReader[]; PolicyDefs.EndOperation[RSReadMail] };
ENDCASE => NULL;
Process.Detach[FORK RegMailWatcher[]];
END;
RegMailWatcher: PROC =
BEGIN
DO RetrieveDefs.WaitForMail[handle];
PolicyDefs.WaitOperation[RSReadMail];
RegMailReader[];
PolicyDefs.EndOperation[RSReadMail];
ENDLOOP;
END;
RegMailReader: PROC =
BEGIN
LockReader[];
DO noMore: BOOLEAN;
state: RetrieveDefs.ServerState;
procs: RetrieveDefs.AccessProcs;
gv: GV RetrieveDefs.AccessProcs;
[noMore, state, procs] ← RetrieveDefs.NextServer[handle];
IF noMore THEN EXIT;
IF state # notEmpty THEN LOOP;
WITH gvProcs:procs SELECT FROM GV => gv←gvProcs; ENDCASE => LOOP;
BEGIN
ENABLE RetrieveDefs.Failed => CONTINUE;
pleaseFlush: BOOLEAN ← TRUE;
DO sender: BodyDefs.RName = [BodyDefs.maxRNameLength];
time: BodyDefs.TimeStamp;
msgExists, archived, deleted: BOOLEAN;
pleaseDelete: BOOLEAN ← TRUE;
[msgExists, archived, deleted] ← gv.nextMessage[handle];
IF deleted THEN LOOP;
IF NOT msgExists THEN EXIT;
gv.startMessage[handle:handle, sender:sender, postmark: @time];
IF CheckAndLog[sender, time]
THEN DO header: BodyDefs.ItemHeader = gv.nextItem[handle];
SELECT header.type FROM
LastItem => EXIT;
updateItem =>
BEGIN
AcceptUpdate[handle, @gv];
EXIT
END;
briefUpdate =>
BEGIN
IF NOT AcceptBriefUpdate[handle, header.length, @gv]
THEN pleaseDelete ← FALSE;
EXIT
END;
ENDCASE => NULL;
ENDLOOP--each item--;
IF pleaseDelete
THEN gv.deleteMessage[handle]
ELSE pleaseFlush ← FALSE;
ENDLOOP--each message--;
IF pleaseFlush THEN gv.accept[handle];
END;
ENDLOOP--each server--;
UnlockReader[];
END;
CheckAndLog: PROC[sender: BodyDefs.RName, time: BodyDefs.Timestamp]
RETURNS [good: BOOLEAN] =
BEGIN
log: STRING = [128];
good ← TRUE;
String.AppendString[log, "RS reading "L];
ProtocolDefs.AppendTimestamp[log, time];
String.AppendString[log, " from "L];
String.AppendString[log, sender];
IF RegServerDefs.IsMember["*.gv"L, sender, direct].membership # yes
THEN {String.AppendString[log, ": bad sender"L]; good ← FALSE};
LogDefs.WriteLogEntry[log];
END;
GetMsgItem: PROCEDURE[handle: RetrieveDefs.Handle,
gv: POINTER TO GV RetrieveDefs.AccessProcs]
RETURNS[ writer: HeapDefs.WriterHandle ] =
BEGIN
buffer: VMDefs.Page = VMDefs.AllocatePage[];
length: CARDINAL;
writer ← HeapDefs.HeapStartWrite[temp];
WHILE (length ← gv.nextBlock[handle, DESCRIPTOR[buffer, 2*VMDefs.pageSize] ]) > 0
DO HeapDefs.HeapWriteData[writer, [buffer, (1+length)/2] ]; ENDLOOP;
VMDefs.Release[buffer];
END;
MangledMessage: ERROR = CODE;
AcceptUpdate: PROCEDURE[handle: RetrieveDefs.Handle,
gv: POINTER TO GV RetrieveDefs.AccessProcs] =
BEGIN
updateWriter: HeapDefs.WriterHandle = GetMsgItem[handle, gv];
AcceptRestOfMessage[handle, gv];
HeapDefs.HeapEndWrite[updateWriter, RegServerDefs.Update];
END; --AcceptUpdate--
AcceptBriefUpdate: PROCEDURE[handle: RetrieveDefs.Handle,
itemLength: BodyDefs.ItemLength,
gv: POINTER TO GV RetrieveDefs.AccessProcs]
RETURNS[ ok: BOOLEAN ] =
BEGIN
strHeader: CARDINAL = 2*SIZE[StringBody[0]];
op: ProtocolDefs.RSOperation;
entry: BodyDefs.RName = [BodyDefs.maxRNameLength];
element: BodyDefs.RName = [BodyDefs.maxRNameLength];
stamp: BodyDefs.Timestamp;
rc: ProtocolDefs.ReturnCode;
length: CARDINAL;
length ← gv.nextBlock[handle,
DESCRIPTOR[LOOPHOLE[@op,POINTER], 2*SIZE[ProtocolDefs.RSOperation]]];
itemLength ← itemLength - length;
length ← gv.nextBlock[handle,
DESCRIPTOR[LOOPHOLE[entry,POINTER], strHeader]];
itemLength ← itemLength - length;
length ← gv.nextBlock[handle,
DESCRIPTOR[LOOPHOLE[@entry.text,POINTER], 2*BodyDefs.RNameSize[entry]-strHeader]];
itemLength ← itemLength - length;
length ← gv.nextBlock[handle,
DESCRIPTOR[LOOPHOLE[element,POINTER], strHeader]];
itemLength ← itemLength - length;
length ← gv.nextBlock[handle,
DESCRIPTOR[LOOPHOLE[@element.text,POINTER],2*BodyDefs.RNameSize[element]-strHeader]];
itemLength ← itemLength - length;
length ← gv.nextBlock[handle,
DESCRIPTOR[LOOPHOLE[@stamp,POINTER], 2*SIZE[BodyDefs.Timestamp]]];
itemLength ← itemLength - length;
IF itemLength # 0 OR length # 2*SIZE[BodyDefs.Timestamp] THEN ERROR MangledMessage[];
AcceptRestOfMessage[handle, gv];
rc ← RegServerDefs.UpdateSublist[entry, element, op, stamp];
RETURN[ rc.type # notFound ]
END;
AcceptRestOfMessage: PROCEDURE[handle: RetrieveDefs.Handle,
gv: POINTER TO GV RetrieveDefs.AccessProcs] =
BEGIN
DO header: BodyDefs.ItemHeader = gv.nextItem[handle];
SELECT header.type FROM
LastItem => EXIT;
ENDCASE => NULL;
ENDLOOP;
END; --AcceptRestOfMessage--
Process.DisableTimeout[@updateCond];
Process.Detach[ FORK SendRSMail[] ];
STOP;
EnableUpdate[];
END.