DIRECTORY
Process USING [ Detach, EnableAborts, SecondsToTicks, SetTimeout ],
Log USING [ ProblemBool, SLOG ],
Nice,
RPC USING [ CallFailed ],
SafeStorage USING [ GetCanonicalType, Type ],
ThParty USING [ Deregister ],
ThPartyMonitorImpl,
ThPartyPrivate
USING [
CFRef, ConversationBody, ConversationData, DestroyConversation, Distribute, DistributionProc, GetEvent, PartyBody, PartyData, recordingRAtom, SmartsData, Verify ],
Thrush USING [ ConversationHandle, ConvEvent, Credentials, Disposition, EventSequence, EventSequenceBody, H, HandleFault, KillHandle, MakeUnique, NB, Priorities, SHHH, StateID, ThHandle ],
Triples USING [ Any, Erase, Foreach, ForeachProc, Select ]
;
CFRef: TYPE = ThPartyPrivate.CFRef;
ConversationData:
TYPE = ThPartyPrivate.ConversationData;
RTConvType: SafeStorage.Type = SafeStorage.GetCanonicalType[CODE[ThPartyPrivate.ConversationBody]];
ConvEvent: TYPE = Thrush.ConvEvent;
Credentials: TYPE = Thrush.Credentials;
H: PROC[r: REF] RETURNS [Thrush.ThHandle] = INLINE { RETURN[LOOPHOLE[Thrush.H[r]]]; };
NB: TYPE = Thrush.NB;
PartyData:
TYPE = ThPartyPrivate.PartyData;
RTPartyType: SafeStorage.Type = SafeStorage.GetCanonicalType[CODE[ThPartyPrivate.PartyBody]];
StateID: TYPE = Thrush.StateID;
Supervision Process . . . One per party
Wait only so long for a call to complete.
Timer:
TYPE =
RECORD [
done: BOOL←FALSE, doneCond: CONDITION, d: Thrush.Disposition←pass ];
IDone: ENTRY PROC[timer: REF Timer] = { NOTIFY timer.doneCond; };
Informer:
PROC[
smarts: ThPartyPrivate.SmartsData, event: ConvEvent, info, yours, last: BOOL, timer: REF Timer] = TRUSTED {
ENABLE
RPC.CallFailed => {
smarts.canProgress←FALSE;
IDone[timer];
ThParty.Deregister[smartsID: H[smarts]];
CONTINUE; };
SELECT (timer.d ← smarts.interface.Progress[shh: smarts.shh, smartsID:
H[smarts],
informationOnly: info, yourParty: yours,
event: event, latestEvent: last ])
FROM
willAlwaysPassThisRequest => smarts.canProgress←FALSE;
ENDCASE;
timer.done←TRUE;
IDone[timer];
};
Supervisor:
PUBLIC PROC[ party: PartyData ] = {
event: ConvEvent;
toDo, failing, latest: BOOL�LSE;
informationOnly: BOOLEAN←FALSE;
Inform: ThPartyPrivate.DistributionProc = {
PROC[smarts: SmartsData ] RETURNS [ d: Thrush.Disposition ];
yourParty: BOOL = event.credentials.partyID=H[party];
d←pass;
IF ~smarts.canProgress THEN RETURN;
IF event.keyTable=
NIL
AND event.intervalSpec=
NIL
AND
(NOT yourParty) AND (event.state#initiating) THEN RETURN; -- don't report uninteresting events
IF ~pd.wdtActive
OR ~smarts.remote
THEN
SELECT (d ← smarts.interface.Progress[shh: smarts.shh, smartsID:
H[smarts],
informationOnly: informationOnly, yourParty: yourParty,
event: event, latestEvent: latest ])
FROM
actedAndPass => informationOnly←TRUE;
willAlwaysPassThisRequest => smarts.canProgress←FALSE;
ENDCASE
ELSE {
timer: REF Timer ← NEW[Timer←[]];
IWait:
ENTRY
PROC
RETURNS [done:
BOOL] =
TRUSTED {
Process.EnableAborts[@timer.doneCond];
Process.SetTimeout[@timer.doneCond, Process.SecondsToTicks[pd.maxNotifyWait]];
Process.Detach[FORK Informer[smarts, event, informationOnly, yourParty, latest, timer]];
WAIT timer.doneCond;
d←timer.d;
IF d=actedAndPass THEN informationOnly ← TRUE;
IF ~timer.done THEN smarts.canProgress ← FALSE;
RETURN[timer.done];
};
IF ~IWait[] THEN ThParty.Deregister[smartsID: H[smarts]];
};
};
TRUSTED {
Process.EnableAborts[@party.actionNeeded];
Process.SetTimeout[@party.actionNeeded, Process.SecondsToTicks[pd.maxSpvrInterval]];
};
DO
ENABLE ABORTED => { party.partyFailed ← TRUE; CONTINUE; };
[ event, toDo, latest ] ← ScanPostOrWait[party];
Have discovered a change in state that our Smarts should know about. The Smarts will now learn about it, even if the state changes again before they can act on the change.
informationOnly ← FALSE;
IF toDo THEN []←ThPartyPrivate.Distribute[party: party, proc: Inform]
ELSE IF failing THEN EXIT
ELSE IF party.partyFailed THEN failing←TRUE -- one more pass to clean up!
ELSE IF party.supervisor=NIL THEN RETURN; -- vanish to keep local frames to minimum.
ENDLOOP;
EndParty[party];
};
Supervise:
PUBLIC
INTERNAL
PROC[party: PartyData] =
TRUSTED {
IF party.supervisor=
NIL
THEN
Process.Detach[party.supervisor ← FORK Supervisor[party]];
NOTIFY party.actionNeeded; -- Spurious if process just spawned?
};
ScanPostOrWait:
ENTRY
PROC[ party: PartyData ]
RETURNS [ event: ConvEvent, toDo: BOOL←FALSE, latest: BOOL ] = {
ENABLE UNWIND => NULL;
Scanner:
INTERNAL Triples.ForeachProc =
TRUSTED {
PROC[trip: TripleRec] RETURNS [continue: BOOLEAN←TRUE]
conv: ConversationData;
WITH trip.att
SELECT
FROM
r: CFRef => {
cfRef: CFRef←NIL;
conv ← NARROW[trip.obj];
IF r.lastNotedID<conv.currentStateID THEN cfRef ← r
ELSE IF r.event.state=idle THEN Dissolve[r, conv, party];
IF cfRef#
NIL
THEN {
cfRef.lastNotedID ← cfRef.lastNotedID+1;
toDo←TRUE;
latest ←
cfRef.lastNotedID = conv.currentStateID OR cfRef.lastNotedID = cfRef.lastPostedID;
event ← ThPartyPrivate.GetEvent[conv, cfRef.lastNotedID];
Log.SLOG[];
RETURN[FALSE];
};
};
ENDCASE;
};
Triples.Foreach[Triples.Any, Triples.Any, party, Scanner];
IF ~toDo
AND ~party.partyFailed
THEN
IF party.numConvs=0 THEN party.supervisor←NIL ELSE WAIT party.actionNeeded;
};
Distribution utility. Basic Party distribution function << At last!! >>
Distribute:
PUBLIC
PROC[ party: PartyData, proc: ThPartyPrivate.DistributionProc ]
RETURNS [ d: Thrush.Disposition←pass ] = {
priorities: Thrush.Priorities;
IF party = NIL THEN RETURN[pass];
priorities ← NARROW[Triples.Select[$Priorities, party, -- priorities -- ]];
Perform distribution in order of this Party's priorities
FOR priorities: Thrush.Priorities ←
NARROW[Triples.Select[$Priorities, party,
--priorities--]], priorities.rest
WHILE priorities#
NIL
DO
<<Flatten next priority level, for now>>
FOR subpriorities: Thrush.Priorities ←
NARROW[priorities.first], subpriorities.rest
WHILE subpriorities#
NIL
DO
priority: ATOM ← NARROW[subpriorities.first];
DistOne: Triples.ForeachProc = {
d ← proc[smarts: NARROW[trip.val]];
SELECT d
FROM
actedAndStop => RETURN[FALSE];
actedAndPass, pass, willAlwaysPassThisRequest => RETURN[TRUE];
ENDCASE=>
RETURN[Log.ProblemBool[
remark: "Invalid distribute return code", where: $System, bool: TRUE]];
};
Distribute to all Smarts at this priority level.
Triples.Foreach[priority, party, Triples.Any, DistOne];
IF d=actedAndStop THEN RETURN; ENDLOOP; ENDLOOP; };
Dissolve:
INTERNAL
PROC[cfRef: CFRef, conv: ConversationData, party: PartyData] = {
If Post generates new work to do, stop and do it, then cycle again.
Post is probably the wrong word for this activity, now. But the work needs doing.
IF cfRef.event.state#idle THEN RETURN;
Triples.Erase[cfRef, conv, party];
conv.numParties ← conv.numParties-1;
party.numConvs ← party.numConvs-1;
IF conv.numParties=0 THEN ThPartyPrivate.DestroyConversation[conv];
When a trunk party leaves a conversation, it loses its identity (<< something wrong here? >>)
When a recording party leaves, it reverts to available.
TRUSTED {
WITH p: party
SELECT
FROM
trunk => { p.outgoing ← NIL; Triples.Erase[$RnameForTrunk, party, Triples.Any]; };
recording => Thrush.MakeUnique[$RnameForParty, party, ThPartyPrivate.recordingRAtom];
ENDCASE;
};
};
EndParty:
ENTRY
PROC[party: PartyData] = {
Thrush.KillHandle[H[party], RTPartyType!Thrush.HandleFault=>CONTINUE];
Triples.Erase[Triples.Any, party, Triples.Any];
party.supervisor ← NIL;
};
GetHistory:
PUBLIC
ENTRY
PROC[
shhh: Thrush.SHHH,
credentials: Credentials,
firstState: StateID,
lastState: StateID -- default: get latest
] RETURNS [ nb: Thrush.NB, events: Thrush.EventSequence←NIL ] = {
ENABLE UNWIND => NULL;
conv: ConversationData;
numStates: NAT;
[conv, , , nb] ← ThPartyPrivate.Verify[credentials];
IF nb#success OR conv=NIL OR conv.currentStateID=0 THEN RETURN;
IF lastState=0 OR lastState>conv.currentStateID THEN lastState𡤌onv.currentStateID;
SELECT firstState
FROM
<1 => firstState𡤁
>conv.currentStateID => RETURN;
ENDCASE;
IF lastState<firstState THEN RETURN;
numStates ← lastState-firstState+1;
events←NEW[Thrush.EventSequenceBody[numStates]];
FOR i: NAT IN [0..numStates) DO events[i] ← conv.log[i+firstState]; ENDLOOP; };
Nice.View[pd, "Party PD"];