<> <> <> DIRECTORY BasicTime, Camelot, CamelotRecoverable, CountedVM, File, FileStream, FS, FSBackdoor, IO, Mach, PBasics, Process, RedBlackTree, Rope, YggBuffMan, YggDID, YggDIDPrivate, YggDIDMap, YggDIDMapPrivate, YggdrasilInit, YggLock, YggLog, YggLogBasic, YggLogControl, YggLogRep, YggFixedNames, YggEnvironment, YggFile, YggFileStream, YggIndexMaint, YggInternal, YggMonitoringLog, YggSunOS, YggRep, YggRestartFile, YggTransaction, VM; MachCamelotEmulationForSunOSImpl: CEDAR MONITOR IMPORTS BasicTime, CamelotRecoverable, CountedVM, FS, IO, PBasics, Process, RedBlackTree, YggBuffMan, YggLog, YggLogBasic, YggLogControl, YggdrasilInit, YggRestartFile, YggTransaction, VM EXPORTS Camelot, Mach, CamelotRecoverable, YggdrasilInit, YggMonitoringLog, YggTransaction = BEGIN OPEN Camelot, Mach; ROPE: TYPE = Rope.ROPE; <> notice: PUBLIC YggMonitoringLog.ProcsRecord _ []; DID: PUBLIC TYPE ~ REF DIDRep; DIDRep: PUBLIC TYPE ~ YggDIDPrivate.DIDRep; Document: TYPE = REF DocumentRep; DocumentRep: PUBLIC TYPE = YggDIDMapPrivate.DocumentRep; TraceLogWriting: BOOL _ TRUE; TraceFileStream: IO.STREAM _ NIL; WordsINWORDS: CARD = WORDS[CARD32]; <> GoAway: BOOL _ FALSE; NextTransCount: CARD _ 10001; CheckPointEphocNumber: CARD _ 1; <> MachCall: PUBLIC SIGNAL [errorCode: msgReturnT, explanation: Rope.ROPE] = CODE; MachAnomaly: PUBLIC SIGNAL [explanation: Rope.ROPE] = CODE; taskSelf: PUBLIC PROC RETURNS [targetTask: taskT] ~ { <> targetTask _ [1]; }; taskNotify: PUBLIC PROC RETURNS [notifyPort: portT] ~ { <> notifyPort _ [2]; }; <<>> <<>> <> vmAllocate: PUBLIC PROC [targetTask: vmTaskT, address: vmAddressT, size: vmSizeT, anywhere: BOOL,raiseSignal: BOOL] RETURNS [mappedAddress: vmAddressT _ 0, kernCode: kernReturnT _ -1] ~ TRUSTED { <> interval: VM.Interval; interval _ VM.Allocate[count: VM.PagesForBytes[size]]; mappedAddress _ LOOPHOLE[VM.AddressForPageNumber[interval.page]]; kernCode _ KernSuccess; }; <<>> VMAllocList: LIST OF AllocItem _ NIL; AllocItem: TYPE = RECORD [ pagingObject: pagingObjectT, offset: vmOffsetT, size: vmSizeT, mappedAddress: vmAddressT, cvmHandle: CountedVM.Handle, allocated: BOOL _ TRUE, dirtied: BOOL _ FALSE, timeOfLastDirty: BasicTime.GMT _ BasicTime.earliestGMT, timeOfLastWrite: BasicTime.GMT _ BasicTime.earliestGMT, dirtiedDuringEphoc: CARD _ 0, ephocOfLastWrite: CARD _ 0, ownedByCheckpointProcess: BOOL _ FALSE ]; vmAllocateWithPager: PUBLIC PROC [targetTask: vmTaskT, address: vmAddressT, size: vmSizeT, anywhere: BOOL, pagingObject: pagingObjectT, offset: vmOffsetT, raiseSignal: BOOL] RETURNS [mappedAddress: vmAddressT _ 0, kernCode: kernReturnT _ -1] ~ TRUSTED { <> [mappedAddress, kernCode] _ vmAllocateWithPagerInner[size, pagingObject, offset, FALSE]; }; <<>> vmAllocateWithPagerInner: ENTRY PROC [size: vmSizeT, pagingObject: pagingObjectT, offset: vmOffsetT, zeroFillOnly: BOOL] RETURNS [mappedAddress: vmAddressT _ 0, kernCode: kernReturnT _ -1] ~ TRUSTED { <> loai: LIST OF AllocItem; <> cvmHandle: CountedVM.Handle; pages: INT _ -1; firstPage: INT _ -1; FOR loai _ VMAllocList, loai.rest UNTIL loai = NIL DO IF loai.first.pagingObject # pagingObject THEN LOOP; IF loai.first.offset > offset+size THEN LOOP; IF offset > loai.first.offset+loai.first.size THEN LOOP; IF loai.first.offset = offset AND loai.first.size = size THEN { IF loai.first.allocated THEN ERROR; EXIT; }; ENDLOOP; IF loai = NIL THEN cvmHandle _ CountedVM.Allocate[words:size/BYTES[WORD]] ELSE { IF loai.first.cvmHandle = NIL THEN cvmHandle _ CountedVM.Allocate[words:size/BYTES[WORD]] ELSE cvmHandle _ loai.first.cvmHandle; }; <> mappedAddress _ LOOPHOLE[cvmHandle.pointer]; pages _ FS.PagesForBytes[size]; IF INT[size] # FS.BytesForPages[pages] THEN ERROR; firstPage _ FS.PagesForBytes[offset]; IF INT[offset] # FS.BytesForPages[firstPage] THEN ERROR; IF loai = NIL THEN { loai _ VMAllocList _ CONS[[pagingObject: pagingObject, offset: offset, size: size, mappedAddress: mappedAddress, cvmHandle: NIL], VMAllocList]; }; IF zeroFillOnly THEN { where: LONG POINTER _ LOOPHOLE[mappedAddress]; nWordsLeft: CARD32 _ size/PBasics.bytesPerWord; WHILE nWordsLeft > 0 DO fillThisTime: CARD32 _ MIN[nWordsLeft, 10000]; PBasics.Fill[where: where, nWords: fillThisTime, value: 0]; where _ where + fillThisTime * UNITS[PBasics.Word]; nWordsLeft _ nWordsLeft - fillThisTime; ENDLOOP; loai.first.allocated _ FALSE; loai.first.mappedAddress _ mappedAddress; loai.first.cvmHandle _ cvmHandle; } ELSE { IF loai.first.cvmHandle = NIL THEN { FS.Read[file: FileForPagingObject[loai.first.pagingObject], from: firstPage, nPages: pages, to: LOOPHOLE[mappedAddress]]; loai.first.mappedAddress _ mappedAddress; loai.first.cvmHandle _ cvmHandle; }; loai.first.allocated _ TRUE; }; kernCode _ KernSuccess; }; <<>> vmDeallocate: PUBLIC ENTRY PROC [targetTask: vmTaskT, address: vmAddressT, size: vmSizeT, raiseSignal: BOOL] RETURNS [kernCode: kernReturnT _ -1] ~ { <> FOR loai: LIST OF AllocItem _ VMAllocList, loai.rest UNTIL loai = NIL DO IF loai.first.mappedAddress = address THEN { pages: INT _ -1; firstPage: INT _ -1; IF size # loai.first.size THEN ERROR; WHILE loai.first.ownedByCheckpointProcess DO Process.Pause[1] ENDLOOP; pages _ FS.PagesForBytes[loai.first.size]; firstPage _ FS.PagesForBytes[loai.first.offset]; FS.Write[file: FileForPagingObject[loai.first.pagingObject], to: firstPage, nPages: pages, from: LOOPHOLE[address]]; <> loai.first.dirtied _ FALSE; loai.first.allocated _ FALSE; loai.first.mappedAddress _ 0; loai.first.cvmHandle _ NIL; EXIT; }; < ERROR;>> ENDLOOP; }; noteDirtyOfMemory: ENTRY PROC [optr: optrT, size: uInt] ~ { po: pagingObjectT; po _ PagingObjectForSegementId[optr.segmentId]; FOR loai: LIST OF AllocItem _ VMAllocList, loai.rest UNTIL loai = NIL DO IF loai.first.pagingObject = po AND (loai.first.offset > optr.lowOffset + size) AND (loai.first.offset + loai.first.size < optr.lowOffset) THEN { } ELSE { loai.first.dirtied _ TRUE; loai.first.timeOfLastDirty _ BasicTime.Now[]; loai.first.dirtiedDuringEphoc _ CheckPointEphocNumber; EXIT; }; REPEAT FINISHED => ERROR; ENDLOOP; }; <> msgSend: PUBLIC PROC [header: REF msgHeaderT, option: msgOptionT, timeout: INT, raiseSignal: BOOL] RETURNS [msgCode: msgReturnT _ -1] ~ { <> ERROR; }; <<>> <<>> msgReceive: PUBLIC PROC [header: REF msgHeaderT, option: msgOptionT, timeout: INT, raiseSignal: BOOL] RETURNS [msgCode: msgReturnT _ -1] ~ { <> <> DO Process.Pause[33]; IF GoAway THEN RETURN[Mach.RcvTimedOut]; ENDLOOP; }; <<>> <> nameServerPort: PUBLIC PROC RETURNS [p: portT] ~ { <> p _ [3]; }; <<>> MachPortsLookup: PUBLIC PROC [targetTask: taskT, raiseSignal: BOOL] RETURNS [intPortSet: portArrayT, intPortArrayCount: INT, kernCode: kernReturnT] ~ { <> xPortArray: REF ARRAY[0..3] OF portT _ NEW[ARRAY[0..3] OF portT _ [[0], [1], [2], [3]]]; intPortArrayCount _ 4; kernCode _ KernSuccess; intPortSet _ LOOPHOLE[xPortArray]; }; <<>> nextPort: portT _ [10]; portAllocate: PUBLIC PROC [targetTask: taskT, raiseSignal: BOOL] RETURNS [newPort: portT, kernCode: kernReturnT _ -1] ~ TRUSTED { <> nextPort _ [nextPort + 1]; newPort _ nextPort; kernCode _ KernSuccess; }; <<>> <<>> portRestrict: PUBLIC PROC [targetTask: taskT, port: portT, raiseSignal: BOOL] RETURNS [kernCode: kernReturnT _ -1] ~ { <> kernCode _ KernSuccess; }; <<>> portUnrestrict: PUBLIC PROC [targetTask: taskT, port: portT, raiseSignal: BOOL] RETURNS [kernCode: kernReturnT _ -1] ~ { <> kernCode _ KernSuccess; }; <<>> <> netnameCheckIn: PUBLIC PROC [ServPort: portT, portName: Rope.ROPE, signature: portT, portId: portT, raiseSignal: BOOL] RETURNS [kernCode: kernReturnT] ~ { <<"check in a name into the local name space">> kernCode _ KernSuccess; }; <<>> <> DSInitialize: PUBLIC PROC [dsPort: portT, raiseSignal: BOOL] RETURNS [serverID: serverIdT, tsPort, mPort, sPort: portT, sharedMemAddr: vmAddressT, seqDescList: ListOfSegmentDesc _ NIL, seqPortList: ListOfPorts, kernCode: Mach.kernReturnT _ -1] ~ TRUSTED { <> serverID _ [1989]; tsPort _ [4]; mPort _ [5]; sPort _ [6]; sharedMemAddr _ 0; seqDescList _ LIST[[serverId: [1234], segmentId: [1066], logicalDisk: 'Z, unused: 'z, highSize: 0, lowSize: RecoverableMemoryPagingSize]]; seqPortList _ LIST[RecoverableMemoryPagingObject]; SegmentIdPagingObjectMap _ LIST[ [[1066], RecoverableMemoryPagingObject, CamelotRecoverable.CamelotRecoverableFile], [YggSunOS.LogSegmentID, LogPagingObject, CamelotRecoverable.CamelotLogFile], [[1492], [9], CamelotRecoverable.RestartFile]]; kernCode _ KernSuccess; }; <<>> CamelotRecoverableInit: PUBLIC PROC RETURNS [seqDescList: Camelot.ListOfSegmentDesc, seqPortList: Mach.ListOfPorts] ~ { seqDescList _ LIST[[serverId: [1234], segmentId: YggSunOS.LogSegmentID, logicalDisk: 'K, unused: 'k, highSize: 0, lowSize: YggSunOS.LogFileSize * YggSunOS.LogBytesPerPage], [serverId: [1234], segmentId: [1492], logicalDisk: 'R, unused: 'r, highSize: 0, lowSize: 4096]]; seqPortList _ LIST[LogPagingObject, [9]]; }; RecoverableMemoryPagingObject: pagingObjectT _ [7]; RecoverableMemoryPagingSize: CARD32 _ 40960000; LogPagingObject: pagingObjectT _ [8]; SegmentIdPagingObjectMap: LIST OF SegmentIdPagingObjectMapItem; SegmentIdPagingObjectMapItem: TYPE = RECORD[ segmentId: segmentIdT, pagingObject: pagingObjectT, backingFile: FS.OpenFile ]; PagingObjectForSegementId: PROC [segmentId: segmentIdT] RETURNS [pagingObject: pagingObjectT] ~ { FOR lospom: LIST OF SegmentIdPagingObjectMapItem _ SegmentIdPagingObjectMap, lospom.rest UNTIL lospom = NIL DO IF lospom.first.segmentId = segmentId THEN RETURN[lospom.first.pagingObject]; REPEAT FINISHED => ERROR ENDLOOP; }; SegementIdForPagingObject: PROC [pagingObject: pagingObjectT] RETURNS [segmentId: segmentIdT] ~ { FOR lospom: LIST OF SegmentIdPagingObjectMapItem _ SegmentIdPagingObjectMap, lospom.rest UNTIL lospom = NIL DO IF lospom.first.pagingObject = pagingObject THEN RETURN[lospom.first.segmentId]; REPEAT FINISHED => ERROR ENDLOOP; }; FileForPagingObject: PROC [pagingObject: pagingObjectT] RETURNS [backingFile: FS.OpenFile] ~ { FOR lospom: LIST OF SegmentIdPagingObjectMapItem _ SegmentIdPagingObjectMap, lospom.rest UNTIL lospom = NIL DO IF lospom.first.pagingObject = pagingObject THEN RETURN[lospom.first.backingFile]; REPEAT FINISHED => ERROR ENDLOOP; }; DSPinObject: PUBLIC PROC [dsPort: portT, tid: tidT, optr: optrT, size: uInt, raiseSignal: BOOL] RETURNS [kernCode: Mach.kernReturnT _ -1] ~ { <> effectiveOptr: optrT _ optr; effectiveSize: uInt _ size; IF BYTES[UNIT] = 2 THEN { effectiveOptr.lowOffset _ PBasics.BITAND[optr.lowOffset, 0FFFFFFFEH]; effectiveSize _ PBasics.BITAND[optr.lowOffset - effectiveOptr.lowOffset + size + 1, 0FFFFFFFEH]; }; rememberPin[tid, effectiveOptr, effectiveSize]; noteDirtyOfMemory[effectiveOptr, effectiveSize]; kernCode _ KernSuccess; }; <<>> OpenTraceFileAndStream: PROC ~ { TraceFileStream _ FS.StreamOpen[fileName: "///Trace/LogTrace.txt", accessOptions: $create, keep: 8, createByteCount: 25600]; }; DSLogNewValue: PUBLIC PROC [dsPort: portT, tid: tidT, optr: optrT, newValue: pointerT, newValueCnt: INT, raiseSignal: BOOL] RETURNS [kernCode: Mach.kernReturnT _ -1] ~ { <> thisRecord: YggLog.RecordID; [thisRecord: thisRecord] _ YggLog.Write[trans: tid, logRecordPhaseType: redo, recordType: writeBytes, optr: optr, recordData: [base: LOOPHOLE[newValue], length: (newValueCnt+3)/4, rest: NIL], force: FALSE]; IF TraceLogWriting AND TraceFileStream = NIL THEN OpenTraceFileAndStream[]; IF TraceLogWriting THEN { p: ENTRY PROC ~ { IO.PutF[TraceFileStream, "New value at %g for seg: %g, offset: %b, byteSize: %g(%b)\n", IO.card[thisRecord.low], IO.card[optr.segmentId.value], IO.card[optr.lowOffset], IO.card[newValueCnt], IO.card[newValueCnt]]; IO.Flush[TraceFileStream]; }; p[]; }; monitoredChangePin[transID: tid, loggingNow: TRUE, forceClear: FALSE, optr: optr, size: newValueCnt]; kernCode _ KernSuccess; }; <<>> NumberOfShortLogRecords: INT _ 0; DSLogOldValueNewValue: PUBLIC PROC [dsPort: portT, tid: tidT, optr: optrT, oldValue: pointerT, oldValueCnt: INT, newValue: pointerT, newValueCnt: INT, raiseSignal: BOOL] RETURNS [kernCode: Mach.kernReturnT _ -1] ~ { <> thisRecord: YggLog.RecordID; <<[] _ YggLog.Write[trans: tid, logRecordPhaseType: undo, recordType: writeBytes, optr: optr, recordData: [base: LOOPHOLE[oldValue], length: (newValueCnt+3)/4, rest: NIL], force: FALSE];>> IF newValueCnt <= 2 THEN { NumberOfShortLogRecords _ NumberOfShortLogRecords + 1; }; [thisRecord: thisRecord] _ YggLog.Write[trans: tid, logRecordPhaseType: redo, recordType: writeBytes, optr: optr, recordData: [base: LOOPHOLE[newValue], length: (newValueCnt+3)/4, rest: NIL], force: FALSE]; IF TraceLogWriting AND TraceFileStream = NIL THEN OpenTraceFileAndStream[]; IF TraceLogWriting THEN { p: ENTRY PROC ~ { IO.PutF[TraceFileStream, "Old/new at %b for seg: %g, offset: %b, byteSize: %g(%b)\n", IO.card[thisRecord.low], IO.card[optr.segmentId.value], IO.card[optr.lowOffset], IO.card[newValueCnt], IO.card[newValueCnt]]; IO.Flush[TraceFileStream]; }; p[]; }; monitoredChangePin[transID: tid, loggingNow: TRUE, forceClear: FALSE, optr: optr, size: oldValueCnt]; kernCode _ KernSuccess; }; <<>> DSQInit: PUBLIC PROC [sharedMemAddr: Mach.vmAddressT] ~ { <> }; <<>> <<>> DSQPreflush: PUBLIC PROC [dsPort: Mach.portT, optr: optrT, sizeInBytes: uInt] ~ { <> po: pagingObjectT; po _ PagingObjectForSegementId[optr.segmentId]; FOR loai: LIST OF AllocItem _ VMAllocList, loai.rest UNTIL loai = NIL DO IF loai.first.pagingObject = po AND loai.first.offset = optr.lowOffset THEN { backingFile: FS.OpenFile; pages: INT _ -1; firstPage: INT _ -1; IF sizeInBytes # loai.first.size THEN ERROR; IF loai.first.cvmHandle = NIL THEN ERROR; pages _ FS.PagesForBytes[loai.first.size]; firstPage _ FS.PagesForBytes[loai.first.offset]; backingFile _ FileForPagingObject[loai.first.pagingObject]; FS.Write[file: backingFile, to: firstPage, nPages: pages, from: LOOPHOLE[loai.first.mappedAddress]]; loai.first.dirtied _ FALSE; <> EXIT; }; REPEAT FINISHED => ERROR; ENDLOOP; }; <<>> <<>> DSQZeroFill: PUBLIC PROC [dsPort: Mach.portT, optr: optrT, sizeInBytes: uInt] ~ { <> po: pagingObjectT; po _ PagingObjectForSegementId[optr.segmentId]; [] _ vmAllocateWithPagerInner[sizeInBytes, po, optr.lowOffset, TRUE]; }; <<>> <> TAAddApplication: PUBLIC PROC [tPort: portT, atPort: portT, authName: Rope.ROPE, raiseSignal: BOOL] RETURNS [applicationID: applicationIdT, taPort: portT, kernCode: Mach.kernReturnT _ -1] ~ TRUSTED { <> kernCode _ KernSuccess; }; <<>> TABegin: PUBLIC PROC [taPort: portT, parentTid: tidT, transType: transactionTypeT, raiseSignal: BOOL] RETURNS [newTid: tidT, kernCode: Mach.kernReturnT _ -1] ~ TRUSTED { <> newTid _ GetNextTrans[]; IF ~noteNewTrans[newTid] THEN ERROR; kernCode _ KernSuccess; }; <<>> TAEnd: PUBLIC PROC [taPort: portT, tid: tidT, protocolType: protocolTypeT, raiseSignal: BOOL] RETURNS [timestamp: timestampT, status: INT, kernCode: Mach.kernReturnT _ -1] ~ TRUSTED { <> thisRecord: YggLog.RecordID; transID: YggTransaction.TransID _ LOOPHOLE[tid]; block: LIST OF YggLog.Block; blockArrayOfWords: LONG POINTER TO ARRAY[0..1024/4) OF CARD32; block _ getScratchBlock[]; block.first.length _ 2; blockArrayOfWords _ LOOPHOLE[block.first.base]; blockArrayOfWords[0] _ tid.top.highTicker; blockArrayOfWords[1] _ tid.top.lowTicker; [thisRecord: thisRecord] _ YggLog.Write[trans: tid, logRecordPhaseType: analysis, recordType: commitTrans, optr: [[0],0,0], recordData: block.first, force: TRUE]; returnScratchBlock[block]; IF TraceLogWriting AND TraceFileStream = NIL THEN OpenTraceFileAndStream[]; IF TraceLogWriting THEN { p: ENTRY PROC ~ TRUSTED { IO.PutF[TraceFileStream, "TAEnd at %g\n", IO.card[thisRecord.low] ]; IO.Flush[TraceFileStream]; }; p[]; }; monitoredChangePin[transID: transID, loggingNow: FALSE, forceClear: FALSE, optr: [[0],0,0] , size: 0]; IF ~removeTrans[transID] THEN ERROR; status _ ErSuccess; kernCode _ KernSuccess; }; <<>> TAKill: PUBLIC PROC [taPort: portT, tid: tidT, status: INT, raiseSignal: BOOL] RETURNS [kernCode: Mach.kernReturnT _ -1] ~ TRUSTED { <> thisRecord: YggLog.RecordID; transID: YggTransaction.TransID _ LOOPHOLE[tid]; block: LIST OF YggLog.Block; blockArrayOfWords: LONG POINTER TO ARRAY[0..1024/4) OF CARD32; block _ getScratchBlock[]; block.first.length _ 2; blockArrayOfWords _ LOOPHOLE[block.first.base]; blockArrayOfWords[0] _ tid.top.highTicker; blockArrayOfWords[1] _ tid.top.lowTicker; YggTransaction.Suspend[transID, ErWaitingTransAborted]; [thisRecord: thisRecord] _ YggLog.Write[trans: tid, logRecordPhaseType: analysis, recordType: abortTrans, optr: [[0],0,0], recordData: block.first, force: TRUE]; <> IF TraceLogWriting AND TraceFileStream = NIL THEN OpenTraceFileAndStream[]; IF TraceLogWriting THEN { p: ENTRY PROC ~ TRUSTED { IO.PutF[TraceFileStream, "TAKill at %g\n", IO.card[thisRecord.low] ]; IO.Flush[TraceFileStream]; }; p[]; }; monitoredChangePin[transID: tid, loggingNow: FALSE, forceClear: FALSE, optr: [[0],0,0] , size: 0]; returnScratchBlock[block]; kernCode _ KernSuccess; }; <<>> <> savedScratchBlocks: LIST OF YggLog.Block _ NIL; getScratchBlock: ENTRY PROC RETURNS [block: LIST OF YggLog.Block] ~ { ENABLE UNWIND => {}; IF savedScratchBlocks # NIL THEN { block _ savedScratchBlocks; savedScratchBlocks _ savedScratchBlocks.rest; } ELSE { interval: VM.Interval; interval _ VM.Allocate[count: VM.PagesForBytes[1024]]; block _ LIST[[base: LOOPHOLE[VM.AddressForPageNumber[interval.page]], length: 1024/4, rest: NIL]]; }; }; returnScratchBlock: ENTRY PROC [block: LIST OF YggLog.Block] ~ { block.rest _ savedScratchBlocks; savedScratchBlocks _ block; }; <> savedScratchTrans: Trans; savedScratchTransForEntries: Trans _ NEW[TransRep]; subtransactionAndTransactionMap: RedBlackTree.Table; checkpointNullTransObj: Trans _ NEW[TransRep]; Trans: TYPE = REF TransRep; TransRep: TYPE = RECORD [ transID: YggEnvironment.TransID, outcome: YggEnvironment.Outcome, latched: BOOL _ FALSE, finishTime: BasicTime.GMT _ BasicTime.nullGMT, suspendTime: BasicTime.GMT _ BasicTime.nullGMT, pageBucketsModified: RECORD[ low32: CARD32, high32: CARD32 ] ]; forABit: CONDITION; getScratchTrans: ENTRY PROC RETURNS [scratchTrans: Trans] ~ { ENABLE UNWIND => {}; IF savedScratchTrans # NIL THEN { scratchTrans _ savedScratchTrans; savedScratchTrans _ NIL; } ELSE { scratchTrans _ NEW[TransRep]; }; }; noteNewTrans: PROC [transID: YggEnvironment.TransID] RETURNS [parentOK: BOOL _ TRUE] ~ { newTrans: Trans; insertTrans: ENTRY PROC ~ { ENABLE UNWIND => {}; data: RedBlackTree.UserData; IF YggTransaction.IsTopLevel[transID] THEN { savedScratchTransForEntries.transID _ transID; data _ RedBlackTree.Lookup[subtransactionAndTransactionMap, savedScratchTransForEntries]; IF data # NIL THEN ERROR; } ELSE ERROR; RedBlackTree.Insert[subtransactionAndTransactionMap, newTrans, newTrans]; }; newTrans _ NEW[TransRep _ [transID: transID, outcome: active, pageBucketsModified: [0, 0] ]]; insertTrans[]; }; StateDuringRecovery: PUBLIC PROC [transID: YggEnvironment.TransID] RETURNS [outcome: YggEnvironment.Outcome] ~ { transFound: BOOL _ FALSE; trans: Trans _ NIL; [transFound, trans] _ FindTrans[transID]; IF transFound THEN { RETURN[commit]; } ELSE RETURN[unknown]; }; FindTrans: ENTRY PROC [transID: YggEnvironment.TransID, setLatch: BOOL _ FALSE] RETURNS [transFound: BOOL _ FALSE, trans: Trans _ NIL] ~ { ENABLE UNWIND => {}; [transFound, trans] _ innerFindTrans[transID, setLatch]; }; innerFindTrans: INTERNAL PROC [transID: YggEnvironment.TransID, setLatch: BOOL _ FALSE] RETURNS [transFound: BOOL _ FALSE, trans: Trans _ NIL] ~ { ENABLE UNWIND => {}; data: RedBlackTree.UserData; IF YggTransaction.IsNullTrans[transID] THEN RETURN [transFound: TRUE, trans: checkpointNullTransObj]; savedScratchTransForEntries.transID _ transID; data _ RedBlackTree.Lookup[subtransactionAndTransactionMap, savedScratchTransForEntries]; IF data = NIL THEN { RETURN[FALSE, NIL]; } ELSE { trans _ NARROW[data]; WHILE trans.latched DO WAIT forABit ENDLOOP; IF setLatch THEN trans.latched _ TRUE; RETURN[TRUE, trans]; }; }; unlatchTrans: ENTRY PROC [trans: Trans] ~ { trans.latched _ FALSE; }; removeTrans: ENTRY PROC [transID: YggEnvironment.TransID] RETURNS [transFound: BOOL _ FALSE] ~ { data: RedBlackTree.UserData; savedScratchTransForEntries.transID _ transID; data _ RedBlackTree.Lookup[subtransactionAndTransactionMap, savedScratchTransForEntries]; IF data = NIL THEN { RETURN[FALSE]; } ELSE { trans: Trans; trans _ NARROW[data]; [] _ RedBlackTree.Delete[subtransactionAndTransactionMap, savedScratchTransForEntries]; RETURN[TRUE]; }; }; <> <> GetKeyProc: RedBlackTree.GetKey = { <> trans: Trans _ NARROW[data]; RETURN[ trans ]; }; CompareProc: RedBlackTree.Compare = { <> dataTrans: Trans _ NARROW[data]; keyTrans: Trans _ NARROW[k]; SELECT keyTrans.transID.bottom.nodeId.value FROM > dataTrans.transID.bottom.nodeId.value => RETURN [greater]; < dataTrans.transID.bottom.nodeId.value => RETURN [less]; ENDCASE => { SELECT keyTrans.transID.bottom.highTicker FROM > dataTrans.transID.bottom.highTicker => RETURN [greater]; < dataTrans.transID.bottom.highTicker => RETURN [less]; ENDCASE => { SELECT keyTrans.transID.bottom.lowTicker FROM > dataTrans.transID.bottom.lowTicker => RETURN [greater]; < dataTrans.transID.bottom.lowTicker => RETURN [less]; ENDCASE => RETURN [equal]; }; }; }; <> pageMask: CARD32 = 077B; noBuckets: CARD = 64; byteMask: CARD32 = 01777B; bytesPerBucket: CARD = 1024; pageBuckets: ARRAY [0..noBuckets) OF LIST OF pinItem; pinItem: TYPE = RECORD[ transID: YggEnvironment.TransID, optr: optrT, size: uInt, pc: pinCode ]; pinCode: TYPE = {pin, pinAndLogged, preventPin}; < it's pinned and maybe updated>> < it's pinned and updated, and the log write has been buffered. A flush of the log will make everything OK.>> < disallow pins for this page>> <<>> hashOptr: PROC [optr: optrT] RETURNS [CARD] ~ { lowPageNo: CARD32; lowPageNo _ PBasics.BITRSHIFT[optr.lowOffset, 10]; RETURN[PBasics.BITAND[lowPageNo, pageMask]]; }; tooLow: BOOL _ FALSE; debugAddress: CARD32 _ 1; debugHits: INT _ 0; rememberPin: ENTRY PROC [transID: YggEnvironment.TransID, optr: optrT, size: uInt, doPreventPin: BOOL _ FALSE] ~ { nowOptr: optrT; sizeLeft: uInt; innerRememberPin: INTERNAL PROC [thisOptr: optrT, bytes: CARD] RETURNS [tryAgain: BOOL _ FALSE] ~ { ENABLE UNWIND => {}; hashPage: CARD; lowPageNo: CARD32; transFound: BOOL _ FALSE; trans: Trans _ NIL; lowPageNo _ PBasics.BITRSHIFT[thisOptr.lowOffset, 10]; hashPage _ hashOptr[thisOptr]; FOR lopi: LIST OF pinItem _ pageBuckets[hashPage], lopi.rest UNTIL lopi = NIL DO loopLowPageNo: CARD32; loopLowPageNo _ PBasics.BITRSHIFT[lopi.first.optr.lowOffset, 10]; IF loopLowPageNo = lowPageNo AND lopi.first.pc = preventPin THEN RETURN[TRUE] ENDLOOP; pageBuckets[hashPage] _ CONS[[transID, thisOptr, bytes, pin], pageBuckets[hashPage]]; [transFound, trans] _ innerFindTrans[transID]; IF ~transFound THEN ERROR; IF hashPage < 32 THEN { trans.pageBucketsModified.low32 _ PBasics.BITOR[PBasics.BITLSHIFT[value: 1, count: hashPage], trans.pageBucketsModified.low32]; } ELSE { trans.pageBucketsModified.high32 _ PBasics.BITOR[PBasics.BITLSHIFT[value: 1, count: hashPage-32], trans.pageBucketsModified.high32]; }; }; nowOptr _ optr; sizeLeft _ size; IF optr.segmentId.value < 50 AND optr.segmentId.value # 0 THEN { tooLow _ TRUE; }; IF optr.lowOffset = debugAddress THEN { debugHits _ debugHits + 1; }; WHILE sizeLeft > 0 DO firstByteOnPage: CARD; bytesThisPage: CARD; firstByteOnPage _ PBasics.BITAND[nowOptr.lowOffset, byteMask]; bytesThisPage _ MIN[sizeLeft, bytesPerBucket - firstByteOnPage]; IF bytesThisPage = 0 OR bytesThisPage > bytesPerBucket THEN ERROR; IF innerRememberPin[nowOptr, bytesThisPage] THEN { WAIT forABit; LOOP; }; sizeLeft _ sizeLeft - bytesThisPage; nowOptr.lowOffset _ nowOptr.lowOffset + bytesThisPage; IF sizeLeft # 0 AND PBasics.BITAND[nowOptr.lowOffset, byteMask] # 0 THEN ERROR; ENDLOOP; }; monitoredChangePin: ENTRY PROC [transID: YggTransaction.TransID, loggingNow: BOOL, optr: optrT, forceClear: BOOL, size: uInt] ~ { changePin[transID, loggingNow, forceClear, optr, size]; }; changePin: INTERNAL PROC [transID: YggTransaction.TransID, loggingNow: BOOL, forceClear: BOOL, optr: optrT, size: uInt] ~ { trans: Trans; transFound: BOOL _ TRUE; lowMod: CARD32; highMod: CARD32; gotIt: BOOL _ FALSE; nowOptr: optrT; sizeLeft: uInt; removePinFromBucket: PROC [bucket: CARD, exactMatch: BOOL, thisOptr: optrT, bytes: CARD] RETURNS [gotOne: BOOL _ FALSE] ~ { prev: LIST OF pinItem _ NIL; FOR lopi: LIST OF pinItem _ pageBuckets[bucket], lopi.rest UNTIL lopi = NIL DO IF YggTransaction.EqualTrans[trans.transID, lopi.first.transID] AND (~loggingNow OR ~exactMatch OR (lopi.first.optr.segmentId = thisOptr.segmentId AND lopi.first.optr.lowOffset = thisOptr.lowOffset)) THEN { IF loggingNow THEN { IF lopi.first.pc # pin THEN LOOP; lopi.first.pc _ pinAndLogged; } ELSE { IF ~forceClear AND lopi.first.pc # pinAndLogged THEN ERROR; IF prev = NIL THEN pageBuckets[bucket] _ lopi.rest ELSE prev.rest _ lopi.rest; }; gotOne _ TRUE; LOOP; }; prev _ lopi; ENDLOOP; }; [transFound, trans] _ innerFindTrans[transID]; IF ~transFound THEN ERROR; IF optr.lowOffset = debugAddress THEN { debugHits _ debugHits + 1; }; IF optr = [[0],0,0] AND size = 0 THEN { gotOne: BOOL _ FALSE; lowMod _ trans.pageBucketsModified.low32; FOR bucktNo: INT IN [0..32) DO IF PBasics.BITAND[lowMod, 1] = 1 THEN { gotOne _ removePinFromBucket[bucktNo, FALSE, optr, size]; gotIt _ gotIt OR gotOne; IF ~gotOne THEN ERROR; }; lowMod _ PBasics.BITRSHIFT[value: lowMod, count: 1]; ENDLOOP; highMod _ trans.pageBucketsModified.high32; FOR bucktNo: INT IN [32..64) DO IF PBasics.BITAND[highMod, 1] = 1 THEN { gotOne _ removePinFromBucket[bucktNo, FALSE, optr, size]; gotIt _ gotIt OR gotOne; IF ~gotOne THEN ERROR; }; highMod _ PBasics.BITRSHIFT[value: highMod, count: 1]; ENDLOOP; IF ~loggingNow THEN trans.pageBucketsModified _ [0 ,0]; } ELSE { nowOptr _ optr; sizeLeft _ size; IF optr.segmentId.value < 50 AND optr.segmentId.value # 0 THEN { tooLow _ TRUE; }; WHILE sizeLeft > 0 DO firstByteOnPage: CARD; bytesThisPage: CARD; bucktNo: INT; gotOne: BOOL _ FALSE; firstByteOnPage _ PBasics.BITAND[nowOptr.lowOffset, byteMask]; bytesThisPage _ MIN[sizeLeft, bytesPerBucket - firstByteOnPage]; IF bytesThisPage = 0 OR bytesThisPage > bytesPerBucket THEN ERROR; bucktNo _ hashOptr[nowOptr]; IF ~(gotOne _ removePinFromBucket[bucktNo, TRUE, nowOptr, bytesThisPage]) THEN ERROR; sizeLeft _ sizeLeft - bytesThisPage; nowOptr.lowOffset _ nowOptr.lowOffset + bytesThisPage; IF sizeLeft # 0 AND PBasics.BITAND[nowOptr.lowOffset, byteMask] # 0 THEN ERROR; ENDLOOP; IF ~loggingNow THEN trans.pageBucketsModified _ [0 ,0]; }; }; <> <<>> CALookup: PUBLIC PROC [nameServerPort: Mach.portT, name: Rope.ROPE, site: Rope.ROPE, numberWanted: INT, maxSeconds: INT, raiseSignal: BOOL] RETURNS [portList: Mach.ListOfPorts, kernCode: Mach.kernReturnT] ~ { <> portList _ LIST[[9]]; kernCode _ KernSuccess; }; <<>> <> GetNextTrans: ENTRY PROC RETURNS [transID: YggTransaction.TransID] ~ { NextTransCount _ NextTransCount + 1; transID _ [top: [lowTicker: NextTransCount], bottom: [lowTicker: NextTransCount]]; }; DoCommit: PROC [transID: YggTransaction.TransID, doCommit: BOOL] ~ { [] _ YggTransaction.Finish[transID, IF doCommit THEN commit ELSE abort]; }; <<>> <> <<>> STServer: PUBLIC PROC [inMsg: REF Camelot.camlibSysReqMsgT, outMsg: REF Camelot.camlibSysRepMsgT] RETURNS [messageUnderstood: BOOL _ FALSE] ~ { }; SRServer: PUBLIC PROC [inMsg: REF Camelot.camlibSysReqMsgT, outMsg: REF Camelot.camlibSysRepMsgT] RETURNS [messageUnderstood: BOOL _ FALSE] ~ { }; ATServer: PUBLIC PROC [inMsg: REF Camelot.camlibSysReqMsgT, outMsg: REF Camelot.camlibSysRepMsgT] RETURNS [messageUnderstood: BOOL _ FALSE] ~ { }; <<>> <> milliSecondsBetweenCheckpoint: INT _ 30000; ephocsBetweenCleans: CARD _ 2; CheckpointProcess: PROC ~ { FOR checkpointCount: INT IN [1 ..INT.LAST] DO <> loai: LIST OF AllocItem; now: BasicTime.GMT; segmentId: segmentIdT; IF checkpointCount > 1 THEN Process.PauseMsec[milliSecondsBetweenCheckpoint]; IF GoAway THEN EXIT; now _ BasicTime.Now[]; CheckPointEphocNumber _ CheckPointEphocNumber + 1; DO findABufferToFlush: ENTRY PROC RETURNS [gotOne: BOOL _ FALSE] ~ { FOR loai _ VMAllocList, loai.rest UNTIL loai = NIL DO IF loai.first.dirtied AND loai.first.allocated AND loai.first.cvmHandle # NIL AND loai.first.pagingObject # LogPagingObject THEN { IF CheckPointEphocNumber - loai.first.ephocOfLastWrite > 2 THEN { loai.first.ownedByCheckpointProcess _ TRUE; RETURN [TRUE]; }; }; ENDLOOP; }; IF ~findABufferToFlush[] THEN EXIT; checkpointNullTransObj.pageBucketsModified _ [0, 0]; segmentId _ SegementIdForPagingObject[loai.first.pagingObject]; rememberPin[transID: YggEnvironment.nullTransID, optr: [segmentId: segmentId, highOffset: 0, lowOffset: loai.first.offset], size: loai.first.size, doPreventPin: TRUE]; { pages: INT _ -1; firstPage: INT _ -1; pages _ FS.PagesForBytes[loai.first.size]; firstPage _ FS.PagesForBytes[loai.first.offset]; FS.Write[file: CamelotRecoverable.CamelotRecoverableFile, to: firstPage, nPages: pages, from: LOOPHOLE[loai.first.mappedAddress]]; loai.first.timeOfLastWrite _ now; loai.first.ephocOfLastWrite _ CheckPointEphocNumber; loai.first.dirtied _ FALSE; loai.first.ownedByCheckpointProcess _ FALSE; }; monitoredChangePin [transID: YggEnvironment.nullTransID, loggingNow: FALSE, forceClear: TRUE, optr: [[0],0,0], size: 0]; ENDLOOP; { checkpointCompleteRecordBody: YggLogRep.CheckpointCompleteRecord _ [startAnalysisRecordID: YggLog.nullRecordID, keepRecordID: YggLog.nullRecordID, checkPointEphocNumber: CheckPointEphocNumber]; thisRecord: YggLog.RecordID; thisWordNumber: YggEnvironment.WordNumber; TRUSTED { checkpointCompleteRecordBody.blockArrayOfWords[0] _ 5910; checkpointCompleteRecordBody.blockArrayOfWords[1] _ 42; [thisRecord: thisRecord, thisWordNumber: thisWordNumber] _ YggLogBasic.Put[ from: [base: @checkpointCompleteRecordBody, length: SIZE[YggLogRep.CheckpointCompleteRecord]/WordsINWORDS], optr: [[0], 0, 0], writeID: TRUE, force: TRUE]; IF TraceLogWriting AND TraceFileStream = NIL THEN OpenTraceFileAndStream[]; IF TraceLogWriting THEN TRUSTED { p: ENTRY PROC ~ TRUSTED { IO.PutF[TraceFileStream, "Checkpoint at %g\n", IO.card[thisRecord.low]]; IO.Flush[TraceFileStream]; }; p[]; }; <<[thisRecord: thisRecord, thisWordNumber: thisWordNumber] _ YggLog.Write[>> <> <> <> <> <> <> <> }; YggRestartFile.WriteRestartRecord[wordNumberForCheckpointCompleteRecord: thisWordNumber, recordIDForCheckpointCompleteRecord: thisRecord]; }; ENDLOOP; }; <> ZeroBackingStore: PROC = { segmentId: segmentIdT; po: pagingObjectT; lowOffset: CARD32 _ 0; cvmHandle: CountedVM.Handle; segmentId _ SegementIdForPagingObject[RecoverableMemoryPagingObject]; po _ PagingObjectForSegementId[segmentId]; cvmHandle _ CountedVM.Allocate[words: 32768/BYTES[WORD]]; TRUSTED {PBasics.Fill[where: LOOPHOLE[cvmHandle.pointer], nWords: 32768/PBasics.bytesPerWord, value: 0];}; WHILE lowOffset < RecoverableMemoryPagingSize DO FS.Write[file: CamelotRecoverable.CamelotRecoverableFile, to: FS.PagesForBytes[lowOffset], nPages: FS.PagesForBytes[32768], from: LOOPHOLE[cvmHandle.pointer]]; lowOffset _ lowOffset + 32768; ENDLOOP; cvmHandle _ NIL; }; ZeroTwoMegsOfBackingStore: PROC = { segmentId: segmentIdT; po: pagingObjectT; lowOffset: CARD32 _ 0; cvmHandle: CountedVM.Handle; sizeToZero: CARD32 _ 1024; sizeToZero _ 2*1024*sizeToZero; segmentId _ SegementIdForPagingObject[RecoverableMemoryPagingObject]; po _ PagingObjectForSegementId[segmentId]; cvmHandle _ CountedVM.Allocate[words: 32768/BYTES[WORD]]; TRUSTED {PBasics.Fill[where: LOOPHOLE[cvmHandle.pointer], nWords: 32768/PBasics.bytesPerWord, value: 0];}; WHILE lowOffset < sizeToZero DO FS.Write[file: CamelotRecoverable.CamelotRecoverableFile, to: FS.PagesForBytes[lowOffset], nPages: FS.PagesForBytes[32768], from: LOOPHOLE[cvmHandle.pointer]]; lowOffset _ lowOffset + 32768; ENDLOOP; cvmHandle _ NIL; }; <> <> <> <> <> <> <> <> <<[mappedAddress: mappedAddress] _ vmAllocateWithPagerInner[4096, po, lowOffset, TRUE];>> <> <<[] _ vmDeallocate[taskSelf[], mappedAddress, 4096, TRUE];>> <> <> <<};>> <<>> <> LogAnalysisProc: YggLogControl.AnalysisProc = { <> <> SELECT type FROM commitTrans => { [] _ noteNewTrans[trans]; }; abortTrans => {}; ENDCASE => ERROR; }; RecoveryBlock: REF ARRAY [0..4096) OF CARD32; LogRecoveryProc: YggLog.RecoveryProc = { <> <> SELECT type FROM writeBytes => { IF FindTrans[trans, FALSE].transFound THEN { status: YggLog.ReadProcStatus; wordsRead: CARDINAL; sizeOfRecordDataInWords: CARD; recordHeader: YggLogRep.TransactionHeader; recordData: LONG POINTER; recordHeader _ YggLogControl.PeekCurrentTransactionHeader[record]; [status: status, wordsRead: wordsRead] _ YggLog.ReadForRecovery[thisRecord: record, wordsToSkip: 0, to: [base: LOOPHOLE[RecoveryBlock], length: 4096]]; IF status = destinationFull THEN ERROR; IF wordsRead < 4 THEN ERROR; -- fix this sizeOfRecordDataInWords _ wordsRead - WORDS[YggLogRep.TransactionHeader]/WORDS[CARD32]; recordData _ LOOPHOLE[RecoveryBlock, LONG POINTER] + UNITS[YggLogRep.TransactionHeader]; <> <> <> <<= [segmentId: segmentIdT, highOffset: CARD16, lowOffset: CARD32]>> TRUSTED {YggBuffMan.WriteBytes[optr: recordHeader.optr, value: recordData, valueCnt: sizeOfRecordDataInWords*BYTES[CARD32]];}; }; }; ENDCASE => ERROR; }; <> Init: PROC = { subtransactionAndTransactionMap _ RedBlackTree.Create[getKey: GetKeyProc, compare: CompareProc]; TRUSTED {Process.InitializeCondition[@forABit, Process.MsecToTicks[10]]; }; RecoveryBlock _ NEW[ARRAY [0..4096) OF CARD32]; YggLogControl.RegisterAnalysisProc[recordType: commitTrans, analysisProc: LogAnalysisProc]; YggLogControl.RegisterAnalysisProc[recordType: abortTrans, analysisProc: LogAnalysisProc]; YggLog.RegisterRecoveryProc[recordType: writeBytes, recoveryProc: LogRecoveryProc]; }; CamelotRecoveryComplete: PUBLIC PROC [] ~ { RecoveryBlock _ NIL; subtransactionAndTransactionMap _ RedBlackTree.Create[getKey: GetKeyProc, compare: CompareProc]; -- create again after recovery to empty out trans seen during recovery TRUSTED {Process.Detach[FORK CheckpointProcess[]]}; YggdrasilInit.RecoveryComplete[]; }; Init[]; END. <<>>