DIRECTORY DelayedFork, FastFork, PeriodicalFork, Process, RuntimeError;

PeriodicalForkImpl: CEDAR MONITOR
IMPORTS FastFork, Process, RuntimeError
EXPORTS DelayedFork, PeriodicalFork ~ 
BEGIN  

forkImmediateTime: INT = 32;
minimumTime: INT = 64;
minimumTicks: Process.Ticks = Process.MsecToTicks[minimumTime];

timeTop: TimeRef м NEW[TimeRec];
TimeRef: TYPE = REF TimeRec;
TimeRec: TYPE = RECORD [
toggle: BOOL м FALSE, --not monitored, there is only 1 PeriodicalProcess
lessFrequent: TimeRef м NIL,
periodicJobs: JobList м NIL,
thisTime: JobList м NIL,
nextTime: JobList м NIL
];

JobRef: TYPE = LIST OF JobRec;
JobList: TYPE = LIST OF JobRec;

JobRec: TYPE = RECORD [
proc: PROC[REF] м NIL, 
data: REF м NIL,
priority: CARD32  м 0,
isForked: ForkState м joined, --prevent forking more than one process
periodic: BOOL м FALSE
];

ForkState: TYPE = {joined, enqueued};

freeList: JobList м NIL;

NewJob: ENTRY PROC RETURNS [JobRef] = INLINE {
RETURN [InternalNewJob[]];
}; 

InternalNewJob: INTERNAL PROC RETURNS [job: JobRef] = INLINE {
IF freeList=NIL 
THEN job м LIST[[]] 
ELSE {
job м freeList; 
freeList м job.rest;
job.rest м NIL;
};
}; 

InternalFreeJob: INTERNAL PROC [job: JobRef] = INLINE {
job.first.data м NIL;
job.rest м freeList; 
freeList м job
}; 

Register: PUBLIC PROC [ms: INT, proc: PROC[REF], data: REF, priority: CARD32] = {
p: PROC[REF] м proc; --trap here if local proc
EntryRegister: ENTRY PROC [ms: INT, proc: PROC[REF], data: REF, priority: CARD32] = {
ref: TimeRef м FindTime[ms];
job: JobRef м InternalNewJob[];
job.first.isForked м joined;
job.first.proc м proc;
job.first.data м data;
job.first.priority м priority;
job.first.periodic м TRUE;
job.rest м ref.periodicJobs;
ref.periodicJobs м job;
};
EntryRegister[ms, p, data, priority];
};

ForkSoon: PUBLIC PROC [ms: INT, proc: PROC[REF], data: REF, priority: CARD32] = {
p: PROC[REF] м proc; --trap here if local proc
EntryDelayedFork: ENTRY PROC [ms: INT, proc: PROC[REF], data: REF] = {
ref: TimeRef м FindTime[ms];
job: JobRef м InternalNewJob[];
job.first.isForked м joined;
job.first.proc м proc;
job.first.data м data;
job.first.priority м priority;
job.first.periodic м FALSE;
job.rest м ref.nextTime;
ref.nextTime м job;
};
IF ms<=forkImmediateTime 
THEN {
job: JobRef м NewJob[];
job.first.isForked м joined;
job.first.proc м proc;
job.first.data м data;
job.first.priority м priority;
job.first.periodic м FALSE;
job.rest м NIL;
Enqueue[job];
IF ms<0 OR pCountHint<=0 THEN FastFork.Fork[Envelope, NIL, internalPriority];
pCountHint м pCountHint-1;
}
ELSE EntryDelayedFork[ms, p, data];
};

FindTime: INTERNAL PROC [ms: INT] RETURNS [ref: TimeRef м timeTop] = INLINE {
WHILE minimumTime<ms DO
IF ref.lessFrequent=NIL THEN ref.lessFrequent м NEW[TimeRec];
ref м ref.lessFrequent;
ms м ms/2;
ENDLOOP;
};

NoOp: PROC [REF] = {};

Unregister: PUBLIC ENTRY PROC [proc: PROC[REF], data: REF] = {
ENABLE UNWIND => NULL;
Match: INTERNAL PROC [job: JobRef] RETURNS [BOOL] = {
RETURN [job.first.proc=proc AND job.first.data=data]
};
UnregisterTime: INTERNAL PROC [time: TimeRef] = {
lag: JobList м NIL;
list: JobList м time.periodicJobs;
WHILE list#NIL DO
IF Match[list] 
THEN {
list.first.proc м NoOp; --so won't be called even if already in delay buffer
IF lag=NIL 
THEN {time.periodicJobs м list.rest; list м time.periodicJobs}
ELSE {lag.rest м list.rest; list м lag.rest}
}
ELSE {lag м list; list м list.rest}; 
ENDLOOP
};
FOR ref: TimeRef м timeTop, ref.lessFrequent WHILE ref#NIL DO
UnregisterTime[ref]
ENDLOOP;
};

unregisterSelf: SIGNAL = CODE;

UnregisterSelf: PUBLIC PROC [] = {
SIGNAL unregisterSelf
};


serializeNum: INT = 64;
serializeYoungest: [0..serializeNum) м 0; 
serializeCnt: INT м 0;
serializeBuffer: REF ARRAY [0..serializeNum) OF JobRef м
 NEW[ARRAY [0..serializeNum) OF JobRef м ALL[NIL]];
delayNum: INT = 128;
delayYoungest: [0..delayNum) м 0; 
delayCnt: INT м 0;
delayBuffer: REF ARRAY [0..delayNum) OF JobRef м
 NEW[ARRAY [0..delayNum) OF JobRef м ALL[NIL]];

pCountHint: INT м 0; --not exact since don't know who waits for conditions.  Not monitored but periodically reset.
countForOne: INT = 5;

ForwardDelayed: PROC [n: INT м 1] = {
WHILE delayCnt>0 --not empty-- AND n>0 DO 
delayCnt м delayCnt-1;
n м n - 1;
BEGIN
job: JobRef м delayBuffer[(delayYoungest+delayNum-delayCnt) MOD delayNum];
Enqueue[job];
END;
ENDLOOP;
};

internalPriority: Process.Priority ~ Process.priorityClient3;

Envelope: PROC [data: REF] = {
job: JobRef м NIL;
pCountHint м pCountHint+countForOne;
DO
job м Dequeue[job];
IF job=NIL THEN {pCountHint м MAX[pCountHint-countForOne, 0]; RETURN};
IF job.first.priority#internalPriority THEN Process.SetPriority[job.first.priority];
job.first.proc[job.first.data ! 
UNWIND => {pCountHint м pCountHint+countForOne; CONTINUE};
unregisterSelf => {Unregister[job.first.proc, job.first.data]; RESUME};
RuntimeError.UNCAUGHT => {pCountHint м pCountHint-countForOne};
];
IF job.first.priority#internalPriority THEN Process.SetPriority[internalPriority];
job.first.isForked м joined;
ENDLOOP
};

Dequeue: ENTRY PROC [last: JobRef] RETURNS [job: JobRef м NIL] = INLINE {
IF last#NIL AND ~last.first.periodic THEN InternalFreeJob[last];
IF serializeCnt>0 --not empty-- THEN {
serializeCnt м serializeCnt-1;
job м serializeBuffer[(serializeYoungest+serializeNum-serializeCnt) MOD serializeNum];
};
};

EntryEnqueue: ENTRY PROC [job: JobRef] RETURNS [success: BOOL] = INLINE {
IF serializeCnt>=serializeNum --full-- THEN RETURN [FALSE];
serializeYoungest м (serializeYoungest+1) MOD serializeNum;
serializeBuffer[serializeYoungest] м job;
serializeCnt м serializeCnt+1;
job.first.isForked м enqueued;
success м TRUE;
};

EnqueueHard: PROC [job: JobRef] = {
pauseLength: INT м 1;
DO 
FastFork.Fork[Envelope, NIL, internalPriority]; 
Process.Pause[pauseLength]; --give it a chance to do some work
IF EntryEnqueue[job].success THEN RETURN;
pauseLength м pauseLength*2;
ENDLOOP;
};

Enqueue: PROC [job: JobRef] = INLINE {
IF ~EntryEnqueue[job].success THEN EnqueueHard[job];
};

Delay: PROC [job: JobRef] = INLINE {
IF delayCnt>=delayNum --full-- THEN ForwardDelayed[3];
delayYoungest м (delayYoungest+1) MOD delayNum;
delayBuffer[delayYoungest] м job;
delayCnt м delayCnt+1;
};

GetTheDelayedForks: ENTRY PROC [time: TimeRef] RETURNS [jobs: JobList] = INLINE {
jobs м time.thisTime;
time.thisTime м time.nextTime;
time.nextTime м NIL; 
};

SweepThisPeriod: PROC [time: TimeRef] = INLINE {
FOR jobs: JobList м time.periodicJobs, jobs.rest WHILE jobs#NIL DO 
IF jobs.first.isForked=joined THEN {
IF time=timeTop THEN Enqueue[jobs] ELSE Delay[jobs]
};
ENDLOOP;
IF time.thisTime#NIL OR time.nextTime#NIL THEN {
jobs: JobList м GetTheDelayedForks[time];
WHILE jobs#NIL DO
next: JobList м jobs.rest; --save the rest field before possible freeing of first job clobbers...
jobs.rest _ NIL;
IF time=timeTop THEN Enqueue[jobs] ELSE Delay[jobs];
jobs м next
ENDLOOP;
};
};

SweepThisAndLargerPeriods: PROC [time: TimeRef] = INLINE {
WHILE time#NIL DO 
SweepThisPeriod[time];
IF time.toggle м ~time.toggle THEN time м time.lessFrequent ELSE RETURN;
ENDLOOP;
};

PeriodicalProcess: PROC [data: REF] = {
DO
SweepThisAndLargerPeriods[timeTop];
IF delayCnt>0 THEN ForwardDelayed[delayCnt/4+1];
pCountHint м 0; 
IF serializeCnt>0 THEN FastFork.Fork[Envelope, NIL, internalPriority];
Process.Pause[minimumTicks];
ENDLOOP
};

FastFork.Fork[PeriodicalProcess, NIL, Process.priorityForeground];
END.

  8  
PeriodicalForkImpl.mesa
Copyright ╙ 1989, 1991, 1992 by Xerox Corporation.  All rights reserved.
Created by Christian Jacobi, March 1, 1989 5:17:26 pm PST
Christian Jacobi, March 5, 1992 11:42 am PST 

--understanding is: only first list element may be used and constitutes job
--understanding is: all elements may be used
--WARNING: job MUST NOT be in any list since rest field used
--don't worry about calling InternalFreeJob; its tricky and occurs rarely
spread the time of forking: 
1) to increase probability of reusing PROCESS resource 
2) to balance load 
Using two buffers
delaybuffer: delays jobs multiples of time intervals 
(so large intervals will spread the calls on multiple minor intervals)
serializeBuffer: serializing calls while minor interval 
(usually will be emptied before next minor interval)
access to "serialize" variables must be monitored

access to "delay" variables not monitored; only one process
--moves "n" jobs from delay buffer into serialize buffer
--execute jobs as long as some are available in serialize buffer
--Returns next job to be executed out of serialize buffer
--last: may be used to free last job [ughh: this saves a monitor entry/exit]
--enqueues job into serialize buffer if space available
--enqueues job into serialize buffer; insists even if spinning necessary
--Fork one more process to empty buffer
--It's ok to loop and fork more processes: buffer is full
--enqueues job into serialize buffer
--enqueues job into delay buffer
--periodic forks
--	not monitored; all what can happen is missing the first turn
--delayed forks
--	monitored; we can't afford missing the only turn
--recurse for double time increment
--old processes which are still alive will probably not help 
--freshly created processes because of overflow need not be considered; we want more parallelism in this case  
 Э╩>  ХNewlineDelimiter
Ц(cedarcode) styleЩ codeЪЬЩKЪЬ
╧eЬ=ЩHKЪЬ9Щ9KЪЬ-Щ-KЩ ЧЪ╧k	Ь>ШGKШ ЧЪ╧nЬЮЬЮШ!KЪЮЬ Ш'KЪЮЬШ&ЧЪЮШKШ ЧKЪЬЮЬШKЪЬЮЬШKЪЬ?Ш?KШ KЪЬЮЬ
Ш KЪЬ	ЮЬЮЬ	ШЪЬ	ЮЬЮЬШKЪЬЮЬЮЬ╧c2ШHKЪЬЮЬШKЪЬЮЬШKЪЬЮЬШKЪЬЮШKШЧKШ ЪЬЮЬЮЬЮЬШKЪаKЩKЧЪЬ	ЮЬЮЬЮЬШKЪЬ,Щ,KШ ЧЪЬЮЬЮЬШKЪ	ЬЮЬЮЬЮЬЮЬШKЪЬЮЬЮШKЪЬ
ЮЬШKЪЬа'ШEKЪЬ
ЮЬЮШKШЧKШ KЪЬЮЬШ%KШ KЪЬЮЬШKШ Ъ
ЯЬЮЬЮЬЮЬЮЬШ.KЪЮЬШKЪЬШKШ ЧЪ
ЯЬЮЬЮЬЮЬЮЬШ>ЪЮЬ
ЮЬШKЪЮЬЮЬШЪЮЬШKЪЬШKЪЬШKЪЬЮЬШKШЧЧKЪЬШЧKШ ЪЯЬЮЬЮЬЮЬШ7KЪЬ<Щ<KЪЬЮЬШKЪЬШKЪЬШKЪЬШЧЪЯЬЮЬЮЬЮЬЮЬЮЬ	ЮЬЮЬШQKЪЬЮЬЮЬ
аШ.ЪЯЬЮЬЮЬЮЬЮЬЮЬ	ЮЬЮЬШUKЪЬШЪЬШKЪЬШKЪЬШKЪЬШKЪЬШKЪЬЮЬШЧKЪЬШKЪЬШKЪЬШЧKЪЬ%Ш%KЪЬШKШ ЧЪЯЬЮЬЮЬЮЬЮЬЮЬ	ЮЬЮЬШQKЪЬЮЬЮЬ
аШ.ЪЯЬЮЬЮЬЮЬЮЬЮЬ	ЮЬШFKЪЬШЪЬШKЪЬШKЪЬШKЪЬШKЪЬШKЪЬЮЬШЧKЪЬШKЪЬШKЪЬШЧЪЮЬШЪЮЬШЪЬШKЪЬШKЪЬШKЪЬШKЪЬШKЪЬЮЬШKЪЬЮЬШЧKЪЬШKЪЮЬЮЬЮЬЮЬШMKЪЬШKЪЬШЧKЪЮЬШ#ЧKЪЬШKШ ЧЪЯЬЮЬЮЬЮЬЮЬЮЬШMЪЮЬЮШKЪЮЬЮЬЮЬЮЬ
Ш=KЪЬШKЪЬ
Ш
KЪЮЬШЧKШKШ ЧKЪЯЬЮЬЮЬШKШ ЪЯ
ЬЮЬЮЬЮЬЮЬЮЬ	ЮЬШ>KЪЮЬЮЬЮЬШЪ
ЯЬЮЬЮЬЮЬЮЬШ5KЪЮЬЮЬШ4KШЧЪЯЬЮЬЮЬШ1KЪЬЮЬШKЪЬ"Ш"ЪЮЬЮЬЮШЪЮЬШЪЮЬШJЪЬIЩIKЪЬа4ШLЪЮЬЮЬШKЪЮЬ:Ш>KЪЮЬ(Ш,ЧKЪЬШЧKЪЮЬ!Ш%ЧKЪЮШЧKШЧЪЮЬ*ЮЬЮЬЮШ=KЪЬШKЪЮЬШЧKШЧKШ KЪЬЮЬЮЬШKШ ЪЯЬЮЬШ"KЪЮЬШKЪЬШЧKШ ЪЬЩKЪЬ&ЮЬ
Щ7KЪЬЩЧЩЩ5KЩFЧЪЬ8Щ8KЪЬ4Щ4ЧЧKШ KЪЬ1Щ1KЪЬЮЬШKЪЬ*Ш*KЪЬЮЬШЪЬЮЬЮЬЮЬ	Ш8KЪЬЮЬЮЬЮЬ
ЮЬЮЬШ3ЧKЩ KЩ;KЪЬ
ЮЬШKЪЬ"Ш"KЪЬ
ЮЬШЪЬЮЬЮЬЮЬ	Ш0KЪЬЮЬЮЬЮЬ
ЮЬЮЬШ/ЧKШ KЪЬЮЬа]ШrKЪЬЮЬШKШ ЪЯЬЮЬЮЬ	Ш%JЩ8ЪЮЬаЬЮЬЮЬШ*KЪЬШKЪЬ
Ш
ЪЮШKЪЬ<ЮЬШJKЪЬШKЪЮЬШЧKЪЮЬШЧKШЧKШ KЪЬ=Ш=KШ ЪЯЬЮЬЮЬШKЪа0ЬЩ@KЪЬЮЬШKЪЬ$Ш$ЪЮШKЪЬШKЪ
ЮЬЮЬЮЬЮЬЮЬШFKЪЮЬ%ЮЬ)ШTЪЬ Ш KЪЮЬ*ЮЬШ:KЪЬ?ЮЬШGKЪЬЮЬ*Ш?KЪЬШЧKЪЮЬ%ЮЬ'ШRKЪЬШKЪЮШЧKШKШ ЧЪЯЬЮЬЮЬЮЬЮЬЮЬШIJЪа)ЬЩ9JЪаЬаFЩLKЪЮЬЮЬЮЬЮЬШ@ЪЮЬаЬЮЬШ&KЪЬШKЪЬDЮЬШVKШЧKЪЬШЧKШ ЪЯЬЮЬЮЬЮЬЮЬЮЬШIJЪЬ7Щ7KЪ
ЮЬаЬЮЬЮЬЮЬШ;KЪЬ*ЮЬШ;KЪЬ)Ш)KЪЬШKЪЬШKЪЬ
ЮЬШKШЧKШ ЪЯЬЮЬШ#JЪЬHЩHKЪЬЮЬШЪЮЬШJЪа'Щ'KЪЬЮЬШ0KЪЬа"Ш>KЪЮЬЮЬЮЬШ)KЪЬШJЪЬ9Щ9KЪЮЬШЧKЪЬШЧKШ ЪЯЬЮЬЮЬШ&JЪЬ$Щ$KЪЮЬЮЬШ4KШKШ ЧЪЯЬЮЬЮЬШ$JЪЬ Щ KЪЮЬаЬЮЬШ6KЪЬ"ЮЬ
Ш/KЪЬ!Ш!KЪЬШKЪЬШЧKШ Ъ
ЯЬЮЬЮЬЮЬЮЬШQKЪЬШKЪЬШKЪЬЮЬШKШЧKШ ЪЯЬЮЬЮЬШ0KЪаЩKЪа?Щ?ЪЮЬ.ЮЬЮЬЮЬШCЪЮЬЮЬШ$KЪЮЬЮЬЮЬШ3KШЧKЪЮЬШЧKЪаЩKЪа3Щ3Ъ
ЮЬЮЬЮЬЮЬЮЬШ0KЪЬ)Ш)ЪЮЬЮЬЮШKЪЬаFШaKЪЬЮЬШKЪЮЬЮЬЮЬШ4KЪЬШKЪЮЬШЧKЪЬШЧKШKШ ЧЪЯЬЮЬЮЬШ:ЪЮЬЮЬЮЬШKЪЬШKЪа#Щ#KЪЮЬЮЬЮЬЮЬШHKЪЮЬШЧKШЧKШ ЪЯЬЮЬЮЬШ'ЪЮШKЪЬ#Ш#KЪЮЬЮЬШ0ЪЬШJЪЬ=Щ=JЪЬoЩoЧKЪЮЬЮЬ0ШFKЪЬШKЪЮШЧKШЧKШ KЪЬ!ЮЬШBKЪЮЬШKШ Ч ЕЧ    ц  0\  