YggWorkerImpl:
CEDAR
MONITOR
LOCKS self
USING self: Handle
IMPORTS
ConstArith,
YggAccessControl,
YggIdentity,
YggImport,
YggTransaction,
BasicTime,
YggClientMap,
YggConversationTable,
ConvertUnsafe,
YggFileControl,
YggFilePageMgr,
YggLock,
YggLockControl,
YggLog,
YggLogControl,
YggLogInline,
YggDummyProcess,
Rope,
YggDummyRPC,
YggMonitoringLog,
YggTransactionMap,
YggWorkerInternal,
VM
EXPORTS
YggTransaction,
YggTransMgr,
YggInternal,
YggMonitoringHooks,
YggTransactionMap,
YggWorkerControl,
YggWorkerInternal
= BEGIN
Conversation: TYPE = YggEnvironment.Conversation;
TransID: TYPE = YggEnvironment.TransID;
nullTransID: TransID = YggEnvironment.nullTransID;
FileStore: TYPE = YggEnvironment.FileStore;
RecordID: TYPE = YggLog.RecordID;
RecordType: TYPE = YggLog.RecordType;
InternalProgrammingError: ERROR = CODE;
Handle: TYPE = YggWorker.Handle;
nullHandle: Handle = YggWorker.nullHandle;
TransObject:
PUBLIC
TYPE = YggWorker.Object;
YggInternal.TransObject
Refused:
PUBLIC
ERROR [why: YggTransMgr.Refusal] =
CODE;
YggTransMgr.Refused.
disableLocalCoordinatorOptimization:
BOOL ←
FALSE;
When TRUE, worker always forces log as if coordinator were remote.
shortWaitOver:
CONDITION;
Client waits here under various unlikely circumstances involving concurrent calls to CreateWorker, PrepareWorker, and FinishWorker, concurrent calls to worker actions that use exclusive mode StartWork, and excessive concurrency among other worker actions. The wait times out after YggWorker.shortWaitTime milliseconds, allowing the situation to be reevaluated. Nobody notifies shortWaitOver.
Parameters to worker log watchdog:
densityFactor:
INT ← 64;
if self.estimatedUpdateCost*densityFactor is larger than the amount of log tied up by this transaction, then the transaction is considered too expensive to abort (unless it is inactive or its log use exceeeds logWordsUsableByClient).
longTimeSinceLastStartWork:
INT ← 1000;
-- seconds (about 15 minutes)
if the time since last start work is greater than this, the the transaction is considered inactive.
logWordsUsableByClient: ConstArith.Const;
absolute upper bound on of log tied up by a transaction.
maxLogWordsForFastRestart: ConstArith.Const -- = ConstArith.FromInt[3000*VM.wordsPerPage] -- ;
fast restart means two minutes: two passes over 3000 pages at 50 pages/sec.
maxLogWordsForWorker: ConstArith.Const;
A worker may comfortably keep this much log active. Beyond this bound, the worker may be aborted (but need not be). These two values are initialized during recovery (based on the log length) and not changed thereafter (except perhaps from the debugger).
Worker-related procs
CreateWorker:
PUBLIC
PROC [
conversation: Conversation, transID: TransID, coordinator: FileStore] = {
! Unknown {coordinator, transID}, OperationFailed {busy};
YggTransaction.CreateWorker.
Called by a client.
The worker object monitor is not held during the call to AlpineCoordinator.RegisterWorker. This was designed to allow AlpineCoordinator.Finish to hold the coordinator object monitor during its calls to workers. (AlpineCoordinator.Finish does not actually work this way).
self: Handle;
alreadyRegistered: BOOL ← FALSE;
registerWorkerResult: YggTransMgr.RegisterWorkerResult;
alpineTransMgr: REF ← NIL;
alpineTransMgr: AlpineTransMgrRpcControl.InterfaceRecord;
conversationOut: Conversation;
IF YggTransactionMap.GetTransHandle[transID] # nullHandle THEN alreadyRegistered ← TRUE
ELSE self ← ConsWorker[transID, YggImport.Register[coordinator]];
IF alreadyRegistered
OR YggTransactionMap.Register[self].alreadyRegistered
THEN {
Drop self on the floor since transID is already registered. We are not allowed to call RegisterWorker; some other process is doing it. Wait for self.state # unknown, then return result.
self ← YggTransactionMap.GetTransHandle[transID];
IF self = nullHandle
OR WaitUntilNotUnknown[self].workerNotActive
THEN
ERROR YggTransaction.Unknown[transID];
}
ELSE {
We registered the volatile worker, so we are responsible to call RegisterWorker and update volatile and permanent data structures to reflect the result.
We cannot return or error until self.state # unknown, lest another process hang in WaitUntilNotUnknown.
conversationOut ← YggConversationTable.Fetch[self.coordinator];
alpineTransMgr ← YggImport.GetTransMgrInterface[self.coordinator];
IF conversationOut = NIL OR alpineTransMgr = NIL THEN GOTO noCoordinator;
registerWorkerResult ← alpineTransMgr.RegisterWorker[conversationOut, transID !
YggDummyRPC.CallFailed => CHECKED { whyCallFailed ← why; GOTO callFailed } ];
EndCreateWorker[self, IF registerWorkerResult = ok THEN create ELSE abort];
EXITS
noCoordinator => {
EndCreateWorker[self, abort];
ERROR YggTransaction.Unknown[coordinator] };
callFailed => {
EndCreateWorker[self, abort];
SELECT whyCallFailed FROM
timeout, unbound => {
self.coordinator.TransMgrInterfaceCallFailed[alpineTransMgr];
alpineTransMgr ← NIL; --must come after CallFailed is unwound
ERROR YggTransaction.Unknown[coordinator] };
busy =>
ERROR YggTransaction.OperationFailed[busy];
runtimeProtocol, stubProtocol => ERROR;
ENDCASE
}
};
};
ConsWorker:
PROC [trans: TransID, coordinator: YggImport.Handle]
RETURNS [Handle] = {
Note: does not establish the locks field of the YggWorker.Object.
RETURN [
NEW[YggWorker.Object ← [
transID: trans,
timeOfLastStartWork: BasicTime.Now[],
coordinator: coordinator,
coordinatorIsRemote: disableLocalCoordinatorOptimization
OR
NOT coordinator.Equal[YggIdentity.myAlpineImportHandle]]]];
};
EndCreateWorker:
ENTRY
PROC [self: Handle, outcome: {create, abort} ] = {
self is registered in YggTransactionMap, in unknown state (no WorkerBegin written).
IF self.state # unknown THEN ERROR;
IF outcome = create
THEN {
LogWorkerBegin[self];
self.state ← active;
IF self.locks = NIL THEN self.locks ← YggLockControl.ConsTransHeader[self];
}
ELSE {
Must change state from unknown since another process in WaitUntilNotUnknown may be waiting for this event.
self.state ← complete;
self.outcome ← abort;
YggTransactionMap.Unregister[self];
};
};
WaitUntilNotUnknown:
ENTRY
PROC [self: Handle]
RETURNS [workerNotActive: BOOL] = {
WHILE self.state = unknown DO WAIT shortWaitOver ENDLOOP;
RETURN [self.state # active];
};
WorkerPrepare:
PUBLIC
PROC [
conversation: Conversation, trans: TransID,
newTrans: YggEnvironment.TransID]
RETURNS [YggTransMgr.WorkerState] = {
! Refused {wrongCoordinator}
YggTransMgr.WorkerPrepare.
Called by the coordinator of "trans".
This procedure is prepared for duplicate calls. Extra callers wait on the global condition variable prepareFinished.
stateAfterPrepare: YggEnvironment.WorkerState;
self: Handle = YggTransactionMap.GetTransHandle[trans];
IF self = nullHandle THEN RETURN [notReady];
IF NOT Rope.Equal[s1: YggClientMap.GetName[conversation], s2: self.coordinator.Name, case: FALSE] THEN ERROR Refused[wrongCoordinator];
SELECT BeginPrepare[self]
FROM
preparing => NULL; -- normal case
ready => RETURN [ready]; -- duplicate call
ENDCASE => RETURN [notReady]; -- duplicate call
We have caused self to enter the preparing state.
IF newTrans # nullTransID
THEN {
Perform specialized version of CreateWorker.
self.continueWorker ← ConsWorker[newTrans, self.coordinator];
IF YggTransactionMap.Register[self.continueWorker].alreadyRegistered
THEN
ERROR;
The newTrans is already known to us. This probably represents an error in the coordinator.
};
{
IF PreventSomeWork[self: self, allowDifficulty: normal].workInProgress
THEN {
Some work is still in progress. Wait for it to go away.
DO
IF ShortWaitForNStarts[self] = 0 THEN EXIT;
ENDLOOP;
};
Call any external caches now; they are allowed to do read/write pages.
YggAccessControl.PhaseOneSpaceAndOwnerChanges[self !
YggAccessControl.LockFailed, YggAccessControl.Unknown => GOTO abort];
IF PreventStartWork[self: self, allowDifficulty: zero].abortThisTrans THEN GOTO abort;
Any call to StartWork for this transaction will now fail.
There is no work in progress, including waiting lock requests. We therefore have exclusive access without entering the monitor.
stateAfterPrepare ← ReadyWorker[self];
EXITS
abort => {
DO
YggLockControl.AbortWaitingRequests[self];
IF ShortWaitForNStarts[self] = 0 THEN EXIT;
ENDLOOP;
stateAfterPrepare ← notReady;
};
};
EndPrepare[self, stateAfterPrepare];
We have caused self to enter either the ready state or the completing state.
SELECT stateAfterPrepare
FROM
notReady => NormalAbortWorker[self];
readOnlyReady => NormalCommitWorker[self];
ready => NULL;
ENDCASE;
RETURN [stateAfterPrepare]
};
BeginPrepare:
ENTRY
PROC [self: Handle]
RETURNS [YggWorker.State] = {
result is # unknown, active.
result is = preparing iff this call caused the transition to enter the preparing state.
DO
SELECT self.state
FROM
unknown, preparing => WAIT shortWaitOver;
ENDCASE => EXIT;
ENDLOOP;
IF self.state = active THEN self.state ← preparing;
RETURN [self.state];
};
ReadyWorker:
PROC [self: Handle]
RETURNS [YggEnvironment.WorkerState] = {
Called during normal operation and during recovery.
The caller has exclusive access to the transaction self, is not holding the monitor.
lockRoutine: Rope.ROPE; -- (name of YggLockControl routine called)
howFailed: YggEnvironment.LockFailure;
readOnly: BOOL;
{readOnly ← YggFileControl.CommitPhase1[self !
YggLock.Failed => {
lockRoutine ← "YggWorkerImpl.ReadyWorker (calling CommitPhase1)";
howFailed ← why;
GOTO lockProblem
};
YggLock.TransAborting => GOTO abort];
YggLockControl.UpgradeLocks[self !
YggLock.Failed => {
lockRoutine ← "YggWorkerImpl.ReadyWorker (calling UpgradeLocks)";
howFailed ← why;
GOTO lockProblem
};
YggLock.TransAborting => GOTO abort];
RETURN [IF readOnly THEN readOnlyReady ELSE ready];
EXITS
lockProblem => {
logProc: PROC [YggMonitoringLog.LockConflictInfo];
IF (logProc ← YggMonitoringLog.notice.lockConflict) #
NIL
THEN
logProc[[
what: howFailed,
where: lockRoutine,
transID: self.transID,
mode: none,
specifics: unknown[],
message: "(sigh) attempted locking mode not known"
]];
RETURN [notReady];
};
abort => RETURN [notReady];
}};
EndPrepare:
ENTRY
PROC [self: Handle,
stateAfterPrepare: YggEnvironment.WorkerState] = {
SELECT stateAfterPrepare
FROM
notReady => {
LogWorkerCompleting[self: self, outcome: abort, force: FALSE];
self.state ← completing;
self.outcome ← abort;
};
readOnlyReady => {
LogWorkerCompleting[self: self, outcome: readOnly, force: FALSE];
self.state ← completing;
self.outcome ← commit;
};
ready => {
LogWorkerReady[self: self, force: self.coordinatorIsRemote];
self.state ← ready;
};
ENDCASE => ERROR;
};
WorkerFinish:
PUBLIC
PROC [
conversation: Conversation, trans: TransID,
requiredOutcome: YggTransMgr.RequiredOutcome] = {
! Refused {wrongCoordinator, notReady};
YggTransMgr.WorkerFinish
Called by the coordinator of "trans".
self: Handle = YggTransactionMap.GetTransHandle[trans];
IF self = nullHandle THEN RETURN; --we must have already finished.
IF
NOT Rope.Equal[
s1: YggClientMap.GetName[conversation], s2: self.coordinator.Name, case: FALSE] THEN
ERROR Refused[wrongCoordinator];
IF requiredOutcome NOT IN [abort .. commit] THEN ERROR;
LocalWorkerFinish[self, requiredOutcome, clientRequest];
};
AbortUnilaterally:
PUBLIC
PROC [
self: Handle, why: YggTransactionMap.AbortReason] = {
YggTransactionMap.AbortUnilaterally
Note some duplicate logic with LocalWorkerFinish.
SELECT BeginFinish[self, abort, returnIfPreparing]
FROM
active => ERROR Refused [notReady];
preparing => {
We cannot determine self's outcome, and are not responsible for making it complete, but we are responsible for ensuring that it leaves the preparing state. Otherwise it might stay there forever, waiting in the lock manager. This loop is really necessary, since neither process involved is synchronized by self's or YggLock's monitors.
DO
YggLockControl.AbortWaitingRequests[self];
IF ShortWaitForState[self] # preparing THEN RETURN;
ENDLOOP;
};
completing => {
We have caused self to enter the completing state, self.outcome = abort.
Hence we are responsible for making it complete.
IF PreventStartWork[self, zero].abortThisTrans
THEN {
Some work is still in progress. Wait for it to go away.
DO
YggLockControl.AbortWaitingRequests[self];
IF ShortWaitForNStarts[self] = 0 THEN EXIT;
ENDLOOP;
};
NormalAbortWorker[self];
temporary kludge, effective since remote coordinators aren't used now.
[] ← YggTransaction.Finish[
YggIdentity.myLocalConversation, self.transID, abort, FALSE];
RETURN;
};
ENDCASE => RETURN;
};
ShortWaitForState:
ENTRY
PROC [self: Handle]
RETURNS [YggWorker.State] = INLINE {
WAIT shortWaitOver;
RETURN [self.state];
};
ShortWaitForNStarts:
ENTRY
PROC [self: Handle]
RETURNS [nStarts: [0..YggWorker.maxStarts]] = INLINE {
WAIT shortWaitOver;
RETURN [self.nStarts];
};
LocalWorkerFinish:
PROC [
self: Handle, requiredOutcome: YggTransMgr.RequiredOutcome,
whyAbort: YggTransactionMap.AbortReason] = {
SELECT BeginFinish[self, requiredOutcome, waitIfPreparing]
FROM
active => ERROR Refused [notReady];
completing => {
We have caused self to enter the completing state, self.outcome = requiredOutcome.
Hence we are responsible for making it complete.
IF requiredOutcome = abort
THEN {
IF PreventStartWork[self, zero].abortThisTrans
THEN {
Some work is still in progress. Wait for it to go away.
DO
YggLockControl.AbortWaitingRequests[self];
IF ShortWaitForNStarts[self] = 0 THEN EXIT;
ENDLOOP;
};
NormalAbortWorker[self]
}
ELSE NormalCommitWorker[self];
};
ENDCASE => RETURN;
};
BeginFinish:
ENTRY
PROC [self: Handle, requiredOutcome: YggTransMgr.RequiredOutcome, option: {returnIfPreparing, waitIfPreparing}]
RETURNS [YggWorker.State] = {
result is # unknown, ready.
result is = active only if requiredOutcome = commit and transaction was unknown or active on entry. (Caller is in error.)
result is = preparing only if returnIfPreparing = TRUE and transaction was preparing on entry. (Caller is trying to perform unilateral abort.)
result is = completing iff this call caused the transaction to enter the completing state. (Caller is now responsible for making transaction complete.)
DO
SELECT self.state
FROM
unknown, completing => WAIT shortWaitOver;
preparing =>
IF option = waitIfPreparing
THEN
WAIT shortWaitOver
ELSE RETURN [preparing];
ENDCASE => EXIT;
ENDLOOP;
IF requiredOutcome = abort
AND (self.state = active
OR self.state = ready)
THEN {
LogWorkerCompleting[self, abort, FALSE];
self.state ← completing;
self.outcome ← abort;
}
ELSE
IF requiredOutcome = commit
AND self.state = ready
THEN {
LogWorkerCompleting[self, commit, self.coordinatorIsRemote];
self.state ← completing;
self.outcome ← commit;
};
RETURN [self.state];
};
NormalCommitWorker:
PROC [self: Handle] = {
self.state = completing, self.outcome = commit
Call any external caches now; they need to know about the commit.
YggAccessControl.PhaseTwoOrAbortSpaceAndOwnerChanges[transHandle: self, outcome: commit, continueTrans: self.continueWorker];
CommitWorker[self];
};
CommitWorker:
PROC [self: Handle] = {
Called during normal operation and during recovery.
YggFileControl.CommitPhase2[trans: self, newTrans: self.continueWorker];
IF self.continueWorker # nullHandle
THEN {
YggLockControl.TransferLocks[from: self, to: self.continueWorker];
self.continueWorker.locks ← self.locks;
self.locks ← NIL;
EndCreateWorker[self.continueWorker, create];
self.continueWorker ← nullHandle;
}
ELSE {
YggLockControl.ReleaseLocks[self];
self.locks ← NIL;
};
};
NormalAbortWorker:
PROC [self: Handle] = {
self.state = completing, self.outcome = abort
Call any external caches now; they need to know about the abort.
YggAccessControl.PhaseTwoOrAbortSpaceAndOwnerChanges[transHandle: self, outcome: abort];
AbortWorker[self];
};
AbortWorker:
PROC [self: Handle] = {
logProc: PROC [YggMonitoringLog.TransactionAbortInfo];
Called during normal operation and during recovery.
IF (logProc ← YggMonitoringLog.notice.abortTransaction) #
NIL
THEN
logProc[[
transID: self.transID,
where: "WorkerImpl.AbortWorker",
locks: self.locks,
why: lockInfo,
message: ""
]];
YggFileControl.Abort[trans: self];
YggLockControl.ReleaseLocks[self];
self.locks ← NIL;
IF self.continueWorker # nullHandle
THEN {
EndCreateWorker[self.continueWorker, abort];
self.continueWorker ← nullHandle;
};
};
Log writing and analysis; checkpoint proc
LogWorkerBegin:
PROC [self: Handle] = {
rec: YggWorker.BeginLogRep ← [];
TRUSTED {
ConvertUnsafe.AppendRope[
to: LOOPHOLE[LONG[@rec.coordinator]], from: self.coordinator.Name];
[thisRecord: self.beginRecord] ← YggLog.Write[self, analysis, workerBegin,
[base: @rec, length: TEXT[rec.length].SIZE]];
};
};
LogWorkerReady:
PROC [self: Handle, force:
BOOL] = {
[] ← YggLog.Write[self, analysis, workerReady,
YggLog.nullBlock, force];
};
LogWorkerCompleting:
PROC [self: Handle,
outcome: YggInternal.WorkerOutcome, force: BOOL] = {
rec: YggWorker.CompletingLogRep ← [outcome: outcome];
TRUSTED {[] ← YggLog.Write[self, analysis, workerCompleting,
[base: @rec, length: YggWorker.CompletingLogRep.SIZE], force]; };
};
LogWorkerComplete:
PROC [self: Handle] = {
[] ← YggLog.Write[self, analysis, workerComplete, YggLog.nullBlock];
};
AnalyzeWorkerBegin:
PROC [
record: RecordID, type: RecordType, trans: TransID] = {
coordinator: YggEnvironment.FileStore;
{
-- get coordinator name from log, map it into a FileStore.
rec: YggWorker.BeginLogRep ← [];
status: YggLog.ReadProcStatus;
TRUSTED {
[status: status] ← YggLog.ReadForRecovery[thisRecord: record,
to: [base: @rec, length: YggWorker.BeginLogRep.SIZE]];
IF status = destinationFull THEN ERROR InternalProgrammingError;
coordinator ← ConvertUnsafe.ToRope[from: LOOPHOLE[LONG[@rec.coordinator]]];
};
};
{
-- create a worker handle from trans, setting its coordinator field from above.
self: Handle ← ConsWorker[trans, YggImport.Register[coordinator]];
IF YggTransactionMap.Register[self].alreadyRegistered THEN ERROR InternalProgrammingError;
self.beginRecord ← record;
self.locks ← YggLockControl.ConsTransHeader[self];
self.allowableDifficulty ← zero;
self.stateDuringRecovery ← active;
};
};
AnalyzeWorkerReady:
PROC [
record: RecordID, type: RecordType, trans: TransID] = {
self: Handle = YggTransactionMap.GetTransHandle[trans];
IF self = nullHandle THEN RETURN;
IF self.stateDuringRecovery # active THEN ERROR InternalProgrammingError;
self.stateDuringRecovery ← ready;
};
AnalyzeWorkerCompleting:
PROC [
record: RecordID, type: RecordType, trans: TransID] = {
outcome: YggInternal.WorkerOutcome;
{ -- get outcome from log
rec: YggWorker.CompletingLogRep;
status: YggLog.ReadProcStatus;
TRUSTED {[status: status] ← YggLog.ReadForRecovery[thisRecord: record,
to: [base: @rec, length: YggWorker.CompletingLogRep.SIZE]]; };
IF status # normal THEN ERROR InternalProgrammingError;
outcome ← rec.outcome;
};
{
self: Handle = YggTransactionMap.GetTransHandle[trans];
IF self = nullHandle THEN RETURN;
IF self.stateDuringRecovery
NOT
IN [active .. ready]
THEN
ERROR InternalProgrammingError;
self.stateDuringRecovery ← IF outcome = abort THEN aborted ELSE committed;
};
};
AnalyzeWorkerComplete:
PROC [
record: RecordID, type: RecordType, trans: TransID] = {
self: Handle = YggTransactionMap.GetTransHandle[trans];
IF self = nullHandle THEN RETURN;
SELECT self.stateDuringRecovery
FROM
active, ready => ERROR;
committed => self.outcome ← commit;
aborted => self.outcome ← abort
ENDCASE => ERROR;
YggLockControl.ReleaseLocks[self];
self.locks ← NIL;
self.state ← complete;
YggTransactionMap.Unregister[self];
};
CallAfterAnalysisPass:
PUBLIC
PROC [] = {
YggWorkerControl.CallAfterAnalysisPass
AbortActiveWorker:
PROC [self: Handle]
RETURNS [stop:
BOOL] = {
SELECT self.stateDuringRecovery
FROM
active => {
LogWorkerCompleting[self: self, outcome: abort, force: FALSE];
self.stateDuringRecovery ← aborted;
};
ready, committed, aborted => NULL;
ENDCASE => ERROR;
RETURN [stop: FALSE];
};
TRUSTED {
logWordsUsableByClient ← ConstArith.FromInt[YggLogControl.WordsUsableByClient[]];
maxLogWordsForWorker ← IF ConstArith.Compare[maxLogWordsForFastRestart, logWordsUsableByClient] = greater THEN maxLogWordsForFastRestart ELSE logWordsUsableByClient;
};
YggTransactionMap.LockedEnumerate[AbortActiveWorker];
};
RecoverWorkerBegin:
PROC [
record: RecordID, type: RecordType, trans: Handle, outcome: YggLog.TransState] = {
IF trans.state # unknown THEN ERROR;
trans.state ← active;
};
RecoverWorkerReady:
PROC [
record: RecordID, type: RecordType, trans: Handle, outcome: YggLog.TransState] = {
IF trans.state # active THEN ERROR;
trans.state ← ready;
};
RecoverWorkerCompleting:
PROC [
record: RecordID, type: RecordType, trans: Handle, outcome: YggLog.TransState] = {
trans.state ← completing;
SELECT outcome
FROM
committed => {
trans.state ← ready;
IF ReadyWorker[trans] = notReady THEN ERROR;
trans.state ← completing; trans.outcome ← commit;
[] ← CommitWorker[trans];
};
aborted => {
trans.state ← completing; trans.outcome ← abort;
AbortWorker[trans];
};
ready => {
trans.state ← ready;
IF ReadyWorker[trans] # ready THEN ERROR;
};
ENDCASE => ERROR;
};
CallAfterUpdatePass:
PUBLIC
PROC [] = {
YggWorkerControl.CallAfterUpdatePass
noActiveWorkers: BOOL;
CheckForActiveWorkers:
ENTRY
PROC [self: Handle]
RETURNS [stop:
BOOL] = {
SELECT self.state
FROM
ready, fpmComplete => NULL;
ENDCASE => noActiveWorkers ← FALSE;
RETURN [stop: FALSE];
};
DO
noActiveWorkers ← TRUE;
YggTransactionMap.LockedEnumerate[CheckForActiveWorkers];
IF noActiveWorkers THEN RETURN ELSE YggDummyProcess.Pause[YggDummyProcess.SecondsToTicks[1]];
ENDLOOP;
};
WorkerCheckpointProc:
PROC []
RETURNS [keepRecord, startAnalysisRecord: RecordID] = {
fpmCompleteWorkerSeen: BOOL ← FALSE;
ExamineWorkerBeforeForceOut:
ENTRY
PROC [self: Handle]
RETURNS [stop:
BOOL] = {
IF self.state = fpmComplete
THEN {
self.state ← fpmCompleteBeingForcedOut;
fpmCompleteWorkerSeen ← TRUE;
};
RETURN [stop: FALSE];
};
oldestBeginRecord: YggLog.RecordID ← YggLog.lastRecordID;
ExamineWorkerAfterForceOut:
ENTRY
PROC [self: Handle]
RETURNS [stop:
BOOL] = {
IF self.state = fpmCompleteBeingForcedOut
THEN {
LogWorkerComplete[self];
YggTransactionMap.Unregister[self];
self.state ← complete;
}
ELSE
IF self.state # unknown
THEN
oldestBeginRecord ← YggLogInline.Min[self.beginRecord, oldestBeginRecord];
RETURN [stop: FALSE];
};
YggTransactionMap.UnlockedEnumerate[ExamineWorkerBeforeForceOut];
IF fpmCompleteWorkerSeen THEN YggFilePageMgr.ForceOutEverything[];
YggTransactionMap.UnlockedEnumerate[ExamineWorkerAfterForceOut];
IF oldestBeginRecord # YggLog.lastRecordID
THEN {
distanceFromOldestBegin: ConstArith.Const;
distanceFromOldestBegin ← YggLogInline.WordsFromSubtract[ YggLogControl.RecordIDOfNextWrite[], oldestBeginRecord];
TRUSTED {IF ConstArith.Compare[distanceFromOldestBegin, oldestBeginRecord] = greater THEN YggWorkerInternal.NotifyLogWatchdog[];};
};
RETURN [oldestBeginRecord, oldestBeginRecord];
};
WorkerLogWatchdogProcess:
PUBLIC
PROC [] = {
transID: TransID ← nullTransID;
w: Handle ← NIL;
wait: BOOL ← FALSE;
DO
IF transID # nullTransID THEN w ← YggTransactionMap.GetTransHandle[transID];
IF w = NIL THEN w ← GetOldestActiveWorker[];
IF w = NIL THEN { transID ← nullTransID; wait ← TRUE }
ELSE
SELECT TestForAbort[w]
FROM
notActive => transID ← nullTransID;
abort => {
logProc: PROC [YggMonitoringLog.TransactionAbortInfo];
IF (logProc ← YggMonitoringLog.notice.abortTransaction) #
NIL
THEN
logProc[[
transID: w.transID,
where: "WorkerImpl.WorkerLogWatchdogProcess",
why: watchDog,
message: "had old uncommitted updates"
]];
AbortUnilaterally[w, oldUncommittedUpdates];
transID ← nullTransID;
};
dontAbort => { transID ← w.transID; wait ← TRUE };
ENDCASE => ERROR;
w ← NIL; -- so that w can be garbage-collected
IF wait THEN { YggWorkerInternal.WaitForNotify[]; wait ← FALSE };
ENDLOOP;
};
TestForAbort:
PUBLIC
ENTRY
SAFE
PROC [self: Handle]
RETURNS [YggMonitoringHooks.ShouldAbort] =
TRUSTED {
logWordsForWorker: YggEnvironment.WordCount =
YggLogInline.WordsFromSubtract[YggLogControl.RecordIDOfNextWrite[], self.beginRecord];
IF self.state # active THEN RETURN [notActive];
IF ConstArith.Compare[logWordsForWorker, maxLogWordsForWorker] = less THEN RETURN [dontAbort];
IF ConstArith.Compare[logWordsForWorker, logWordsUsableByClient] = greater THEN RETURN [abort];
IF BasicTime.Period[from: self.timeOfLastStartWork, to: BasicTime.Now[]] <
longTimeSinceLastStartWork --active-- AND
self.estimatedUpdateCost * densityFactor > ConstArith.ToInt[logWordsForWorker] --expensive-- THEN
RETURN [dontAbort];
RETURN [abort];
};
InactiveWorker: PUBLIC ENTRY SAFE PROC [self: Handle] RETURNS [BOOLEAN] ~ CHECKED {
"Inactive" = haven't done a "StartWork" for some number of seconds.
RETURN [BasicTime.Period[from: self.timeOfLastStartWork, to: BasicTime.Now[]] > longTimeSinceLastStartWork];
};
GetOldestActiveWorker:
PROC []
RETURNS [Handle] = {
w: Handle ← NIL;
ExamineWorker:
ENTRY
PROC [self: Handle]
RETURNS [stop:
BOOL] = {
IF self.state = active
AND (w =
NIL
OR
YggLogInline.Compare[w.beginRecord, self.beginRecord] # less) THEN
w ← self;
RETURN [stop: FALSE];
};
YggTransactionMap.UnlockedEnumerate[ExamineWorker];
RETURN [w];
};
Procedures exported to YggTransactionMap, and some related procedures.
StartWork:
PUBLIC
ENTRY
PROC [self: Handle, difficulty: YggInternal.WorkLevel]
RETURNS [canWork:
BOOL] = {
IF difficulty > self.allowableDifficulty THEN RETURN [FALSE];
WHILE self.state = unknown DO WAIT shortWaitOver ENDLOOP;
IF self.state NOT IN [active .. preparing] THEN RETURN [FALSE];
WHILE self.nStarts = YggWorker.maxStarts
DO
WAIT shortWaitOver;
IF difficulty > self.allowableDifficulty THEN RETURN [FALSE];
ENDLOOP;
self.nStarts ← self.nStarts + 1;
self.timeOfLastStartWork ← BasicTime.Now[];
RETURN [TRUE];
};
StopWork:
PUBLIC
ENTRY
PROC [self: Handle] = {
IF self.nStarts = 0 THEN ERROR; --impossible if clients are correctly coded--
self.nStarts ← self.nStarts - 1;
};
PreventStartWork:
ENTRY
PROC [self: Handle, allowDifficulty: YggWorker.Difficulty]
RETURNS [abortThisTrans: BOOL] = {
Return abortThisTrans ~ TRUE if work is in progress.
IF self.nStarts # 0
THEN {
self.allowableDifficulty ← zero;
RETURN [abortThisTrans: TRUE];
}
ELSE {
self.allowableDifficulty ← allowDifficulty;
RETURN [abortThisTrans: FALSE];
};
};
PreventSomeWork:
ENTRY
PROC [self: Handle, allowDifficulty: YggWorker.Difficulty]
RETURNS [workInProgress: BOOL] = {
self.allowableDifficulty ← allowDifficulty;
Return workInProgress ~ TRUE if work is in progress.
RETURN [workInProgress: self.nStarts # 0];
};
GetTimeOfLastStartWork:
PUBLIC
ENTRY
PROC [self: Handle]
RETURNS [BasicTime.GMT] = {
RETURN [self.timeOfLastStartWork];
};
GetEstimatedUpdateCost:
PUBLIC
ENTRY
PROC [self: Handle]
RETURNS [
INT] = {
RETURN [self.estimatedUpdateCost];
};
GetTransID:
PUBLIC
PROC [self: Handle]
RETURNS [TransID] = {
Not ENTRY because the trans field is immutable.
RETURN [self.transID];
};
GetOpenDocList:
PUBLIC
PROC [self: Handle]
RETURNS [openDocList: YggInternal.OpenDoc] = {
Not ENTRY because self.openDocList "belongs" to the open doc manager.
RETURN [self.openDocList];
};
SetOpenDocList:
PUBLIC
PROC [self: Handle,
documentList: YggInternal.OpenDoc] = {
Not ENTRY because self.openDocList "belongs" to the open doc manager.
self.openDocList ← documentList;
};
GetLockHeader:
PUBLIC
PROC [self: Handle]
RETURNS [lockHeader: YggInternal.LockTransHeader] = {
Not ENTRY because self.locks "belongs" to the lock manager.
RETURN[self.locks];
};
EnableAlpineWheel:
PUBLIC
ENTRY
PROC [self: Handle,
conversation: Conversation, enable: BOOL] = {
l: LIST OF YggDummyRPC.ConversationID;
c: YggDummyRPC.ConversationID = YggDummyRPC.GetConversationID[conversation];
IF c = YggIdentity.myLocalConversationID THEN RETURN;
FOR l ← self.enabledWheelList, l.rest
UNTIL l =
NIL
DO
IF l.first = c THEN EXIT;
ENDLOOP;
IF enable
AND l =
NIL
THEN {
new:
LIST
OF
YggDummyRPC.ConversationID =
CONS[
first: c, rest: self.enabledWheelList];
self.enabledWheelList ← new;
}
ELSE
IF (
NOT enable)
AND l #
NIL
THEN {
IF self.enabledWheelList = l THEN self.enabledWheelList ← l.rest
ELSE
FOR p:
LIST
OF
YggDummyRPC.ConversationID ← self.enabledWheelList, p.rest
UNTIL p =
NIL
DO
IF p.rest = l THEN { p.rest ← l.rest; EXIT };
ENDLOOP;
};
};
IsAlpineWheel:
PUBLIC
ENTRY
PROC [self: Handle,
conversation: Conversation] RETURNS [enabled: BOOL] = {
c: YggDummyRPC.ConversationID = YggDummyRPC.GetConversationID[conversation];
IF c = YggIdentity.myLocalConversationID THEN RETURN [enabled: TRUE];
FOR l:
LIST
OF
YggDummyRPC.ConversationID ← self.enabledWheelList, l.rest
UNTIL l =
NIL
DO
IF l.first = c THEN RETURN [enabled: TRUE];
ENDLOOP;
RETURN [enabled: FALSE];
};
StateDuringRecovery:
PUBLIC
ENTRY
PROC [self: Handle]
RETURNS [YggTransactionMap.TransState] = {
SELECT self.stateDuringRecovery
FROM
active => ERROR;
ready => RETURN [ready];
committed => RETURN [committed];
aborted => RETURN [aborted];
ENDCASE => ERROR;
};
IsCommitted:
PUBLIC
ENTRY
PROC [self: Handle]
RETURNS [
BOOL] = {
SELECT self.state
FROM
active, preparing => RETURN [FALSE];
completing => RETURN [self.outcome = commit];
ENDCASE => ERROR; -- locks are supposed to prevent this
};
AssertUpdatesAreComplete:
PUBLIC
ENTRY
PROC [self: Handle] = {
IF self.state # completing THEN ERROR;
self.state ← fpmComplete;
};
TransIDFromTransHandle:
PUBLIC
SAFE
PROC [h: YggInternal.TransHandle]
RETURNS [YggEnvironment.TransID] ~
CHECKED {
For YggMonitoringHooks.
wh: YggWorker.Handle;
IF h =
NIL
THEN
RETURN [YggEnvironment.nullTransID]
ELSE {
TRUSTED {wh ← LOOPHOLE[h]};
RETURN [wh.transID]
}
};
YggLogControl.RegisterAnalysisProc[workerBegin, AnalyzeWorkerBegin];
YggLogControl.RegisterAnalysisProc[workerReady, AnalyzeWorkerReady];
YggLogControl.RegisterAnalysisProc[workerCompleting, AnalyzeWorkerCompleting];
YggLogControl.RegisterAnalysisProc[workerComplete, AnalyzeWorkerComplete];
YggLog.RegisterRecoveryProc[workerBegin, RecoverWorkerBegin];
YggLog.RegisterRecoveryProc[workerReady, RecoverWorkerReady];
YggLog.RegisterRecoveryProc[workerCompleting, RecoverWorkerCompleting];
YggLogControl.RegisterCheckpointProc[WorkerCheckpointProc];
TRUSTED {
maxLogWordsForFastRestart ← ConstArith.FromInt[3000*VM.wordsPerPage];
YggDummyProcess.SetTimeout[@shortWaitOver, YggDummyProcess.MsecToTicks[YggWorker.shortWaitTime]];
YggDummyProcess.EnableAborts[@shortWaitOver];
};
END.--WorkerImpl