"
COPYRIGHT (c) 1993 by Claus Gittinger
All Rights Reserved
This software is furnished under a license and may be used
only in accordance with the terms of that license and with the
inclusion of the above copyright notice. This software may not
be provided or otherwise made available to, or used by, any
other person. No title to or ownership of the software is
hereby transferred.
"
Object subclass:#ProcessorScheduler
instanceVariableNames:'quiescentProcessLists scheduler
zombie
activeProcess currentPriority
readFds readSemaphores readChecks
writeFds writeSemaphores
timeouts timeoutActions timeoutProcesses timeoutSemaphores
idleActions anyTimeouts dispatching'
classVariableNames:'KnownProcesses KnownProcessIds
PureEventDriven
UserSchedulingPriority
UserInterruptPriority
TimingPriority
SchedulingPriority'
poolDictionaries:''
category:'Kernel-Processes'
!
ProcessorScheduler comment:'
COPYRIGHT (c) 1993 by Claus Gittinger
All Rights Reserved
$Header: /cvs/stx/stx/libbasic/Attic/ProcSched.st,v 1.19 1994-08-22 12:11:30 claus Exp $
'!
Smalltalk at:#Processor put:nil!
!ProcessorScheduler class methodsFor:'documentation'!
copyright
"
COPYRIGHT (c) 1993 by Claus Gittinger
All Rights Reserved
This software is furnished under a license and may be used
only in accordance with the terms of that license and with the
inclusion of the above copyright notice. This software may not
be provided or otherwise made available to, or used by, any
other person. No title to or ownership of the software is
hereby transferred.
"
!
version
"
$Header: /cvs/stx/stx/libbasic/Attic/ProcSched.st,v 1.19 1994-08-22 12:11:30 claus Exp $
"
!
documentation
"
This class has only one instance, which is bound to the global
'Processor'. It is responsible for scheduling among the smalltalk
processes (threads; not to confuse with heavy weight unix processes).
Scheduling is fully done in smalltalk (the always runnable scheduler-
process, running at highest priority does this).
The main primitive support is used in threadSwitch, which passes
control to another process (usually selected by the scheduler).
Thus it is possible to modify the schedulers policy.
(To answer a frequently asked question:
dont add preemtive round-robin here; this can be implemented without
any need to change the scheduler. See goodies/timeslicing.st for how
this is done in a very elegant way).
Notice: Smalltalk/X can (still) be compiled & configured without
process support. This non-process mode is called 'pureEventDriven' mode
and is useful to quickly port ST/X to systems, where these facilities
are either not needed (server applications), or are difficult to
implement (threads require some assembler support functions).
To allow pureEvent mode, kludges are built into some places in the
system, where either a process is forked, or a timeout is used instead
(for examples, see ProcessMonitor or MemoryMonitor).
This pure-event mode may not be supported in the future.
class variables:
KnownProcesses <Collection> all known processes
KnownProcessIds <Collection> and their IDs
PureEventDriven <Boolean> true, if no process support
is available
UserSchedulingPriority <Integer> the priority at which normal
user interfaces run
UserInterruptPriority the priority at which user-
interrupts (Cntl-C) processing
takes place. Processes with
a greater or equal priority are
not interruptable.
TimingPriority the priority used for timing.
Processes with a greater or
equal priority are not interrupted
by timers.
SchedulingPriority The priority of the scheduler (must
me higher than any other).
most interresting methods:
Processor>>suspend: (see also Process>>suspend)
Processor>>resume: (see also Process>>resume)
Processor>>terminate: (see also Process>>terminate)
Processor>>yield
Processor>>changePriority:for: (see also Process>>priority:
Processor>>signal:afterSeconds: (see also Delay>>forSeconds:)
Processor>>signal:afterMilliseconds: (see also Delay>>forMilliseconds:)
Processor>>signal:onInput: (see also ExternalStream>>readWait)
Processor>>signal:onOutput: (see also ExternalStream>>writeWait)
Processor>>disableSemaphore:
"
! !
!ProcessorScheduler class methodsFor:'initialization'!
initialize
"class setup: create the one-and-only instance of myself and
setup some priority values."
UserSchedulingPriority := 8.
UserInterruptPriority := 24.
TimingPriority := 16.
SchedulingPriority := 31.
KnownProcesses isNil ifTrue:[
KnownProcesses := WeakArray new:10.
KnownProcesses watcher:self.
KnownProcessIds := OrderedCollection new.
"want to get informed when returning from snapshot"
ObjectMemory addDependent:self
].
Processor isNil ifTrue:[
"create the one and only processor"
Processor := self new.
].
"
allow configurations without processes
"
PureEventDriven := self threadsAvailable not.
PureEventDriven ifTrue:[
'no process support - running event driven' errorPrintNL
].
!
update:something
"being a dependent of the ObjectMemory, this is the notification
that something happened"
something == #restarted ifTrue:[
self reinstallProcesses
]
!
reinstallProcesses
"recreate all processes after a snapShot load.
This is currently not implemented (and might never be).
All we could do is to restart the processes. Time will show."
KnownProcesses do:[:p |
p notNil ifTrue:[
"how, exactly should this be done ?"
p id ~~ 0 ifTrue:[
'process restart not implemented' errorPrintNL
]
]
]
! !
!ProcessorScheduler class methodsFor:'instance creation'!
new
"there is (currently) only one processor ..."
Processor isNil ifTrue:[
Processor := self basicNew initialize
].
^ Processor.
! !
!ProcessorScheduler class methodsFor:'instance release'!
informDispose
"some Process has been collected - terminate the underlying thread"
|id sz "{ Class: SmallInteger }"|
sz := KnownProcessIds size.
1 to:sz do:[:index |
(KnownProcesses at:index) isNil ifTrue:[
id := KnownProcessIds at:index.
id notNil ifTrue:[
Transcript showCr:('terminate thread ',
id printString,
' (no longer refd)').
self threadDestroy:id.
KnownProcessIds at:index put:nil.
]
]
]
! !
!ProcessorScheduler class methodsFor:'queries'!
isPureEventDriven
"this is temporary - (maybe not :-).
you can run ST/X either with or without processes.
Without, there is conceptionally a single process handling all
outside events and timeouts. This has some negative implications
(Debugger is ugly), but allows a fully portable ST/X without any
assembler support - i.e. quick portability.
The PureEvent flag will automatically be set if the runtime system
does not support threads - otherwise, it can be set manually
(from rc-file).
"
^ PureEventDriven
!
pureEventDriven
"turn on pure-event driven mode - no processes, single dispatch loop"
PureEventDriven := true
!
processDriven
"turn on process driven mode"
PureEventDriven := false
!
knownProcesses
"return a collection of all (living) processes in the system"
^ KnownProcesses select:[:p | p notNil]
! !
!ProcessorScheduler class methodsFor:'primitive process primitives'!
threadsAvailable
"return true, if the runtime system supports threads (i.e. processes);
false otherwise."
%{ /* NOCONTEXT */
extern OBJ __threadsAvailable();
RETURN (__threadsAvailable());
%}
!
threadInterrupt:id
"make the process evaluate an interrupt"
%{ /* NOCONTEXT */
if (_isSmallInteger(id)) {
__threadInterrupt(_intVal(id));
}
%}
!
threadCreate:aBlock
"physical creation of a process executing aBlock.
(warning: low level entry, no administration done).
This may return nil, if process could not be created."
%{ /* NOCONTEXT */
int tid;
extern int __threadCreate();
tid = __threadCreate(aBlock, 0 /* stackSize no longer needed */);
if (tid != 0) {
RETURN ( _MKSMALLINT(tid));
}
%}
.
"
arrive here, if creation of process in VM failed.
(no memory for process)
"
^ ObjectMemory allocationFailureSignal raise.
!
threadDestroy:id
"physical destroy other process ...
(warning: low level entry, no administration done)"
%{ /* NOCONTEXT */
if (_isSmallInteger(id)) {
__threadDestroy(_intVal(id));
}
%}
! !
!ProcessorScheduler methodsFor:'primitive process primitives'!
threadSwitch:aProcess
"continue execution in aProcess.
(warning: low level entry, no administration is done here)"
|id pri ok oldProcess oldPri p singleStep wasBlocked|
aProcess isNil ifTrue:[^ self].
aProcess == activeProcess ifTrue:[^ self].
wasBlocked := OperatingSystem blockInterrupts.
oldProcess := activeProcess.
oldPri := currentPriority.
id := aProcess id.
pri := aProcess priority.
singleStep := aProcess isSingleStepping.
aProcess state:#active.
oldProcess state == #active ifTrue:[
oldProcess state:#run.
].
"no interrupts now - activeProcess has already been changed
(dont add any message sends here)"
activeProcess := aProcess.
currentPriority := pri.
%{
extern OBJ __threadSwitch(), __threadSwitchWithSingleStep();
if (singleStep == true)
ok = __threadSwitchWithSingleStep(__context, _intVal(id));
else
ok = __threadSwitch(__context, _intVal(id));
%}.
"time passes ...
... here again"
ok ifFalse:[
"
switch failed for some reason -
destroy the bad process
"
p := activeProcess.
activeProcess := oldProcess.
currentPriority := oldPri.
p id ~~ 0 ifTrue:[
p state:#suspended.
p terminate.
]
].
zombie notNil ifTrue:[
self class threadDestroy:zombie.
zombie := nil
].
wasBlocked ifFalse:[OperatingSystem unblockInterrupts].
!
scheduleForInterrupt:aProcess
"make aProcess evaluate its pushedInterrupt block(s)"
|id|
aProcess isNil ifTrue:[^ self].
aProcess == activeProcess ifTrue:[^ self].
id := aProcess id.
self class threadInterrupt:id.
"
and, make the process runnable
"
aProcess resume
! !
!ProcessorScheduler methodsFor:'constants'!
lowestPriority
"return the lowest priority value"
^ 1 "do not change this - its not variable"
!
highestPriority
"return the highest priority value processes can have"
"must be below schedulingPriority -
otherwise scheduler could be blocked ...
"
^ SchedulingPriority - 1
!
schedulingPriority
"return the priority at which the scheduler runs"
"must be above highestPriority -
otherwise scheduler could be blocked ...
"
^ SchedulingPriority
!
userInterruptPriority
"return the priority, at which the event scheduler runs - i.e.
all processes running at alower priority are interruptable by Cntl-C
or the timer."
^ UserInterruptPriority
!
timingPriority
"return the priority, at which all timing takes place (messageTally,
delay etc.)"
^ TimingPriority
!
userSchedulingPriority
"return the priority, at which all normal user (interactive) processing
takes place"
^ UserSchedulingPriority
!
userBackgroundPriority
"not currently used - for ST80 compatibility only"
^ 6
!
systemBackgroundPriority
"not currently used - for ST80 compatibility only"
^ 4
!
lowIOPriority
"not currently used - for ST80 compatibility only"
^ 2 "claus: is this ok ?"
! !
!ProcessorScheduler methodsFor:'private initializing'!
initialize
"initialize the one-and-only ProcessorScheduler"
|nPrios l p|
nPrios := SchedulingPriority.
quiescentProcessLists := Array new:nPrios.
readFds := Array with:nil.
readChecks := Array with:nil.
readSemaphores := Array with:nil.
writeFds := Array with:nil.
writeSemaphores := Array with:nil.
timeouts := Array with:nil.
timeoutSemaphores := Array with:nil.
timeoutActions := Array with:nil.
timeoutProcesses := Array with:nil.
anyTimeouts := false.
dispatching := false.
"
handcraft the first (dispatcher-) process - this one will never
block, but go into a select if there is nothing to do.
Also, it has a prio of max+1 - thus, it comes first when looking
for a runnable process.
"
currentPriority := SchedulingPriority.
p := Process new.
p setId:0.
p name:'scheduler'.
p state:#run.
p setPriority:currentPriority.
l := LinkedList new.
l add:p.
scheduler := activeProcess := p.
quiescentProcessLists at:currentPriority put:l.
ObjectMemory ioInterruptHandler:self.
ObjectMemory timerInterruptHandler:self.
!
reInitialize
"all previous stuff is obsolete - each object should reinstall itself
upon restart."
KnownProcesses := WeakArray new:5.
KnownProcesses watcher:self class.
KnownProcessIds := OrderedCollection new.
self initialize
! !
!ProcessorScheduler methodsFor:'private'!
remember:aProcess
"remember aProcess for later disposal (where the underlying
system resources have to be freed)."
|newShadow oldSize oldId wasBlocked
index "{ Class: SmallInteger }"
sz "{ Class: SmallInteger }" |
wasBlocked := OperatingSystem blockInterrupts.
index := 1.
sz := KnownProcessIds size.
[index <= sz] whileTrue:[
(KnownProcesses at:index) isNil ifTrue:[
oldId := KnownProcessIds at:index.
oldId notNil ifTrue:[
self class threadDestroy:oldId.
].
KnownProcesses at:index put:aProcess.
KnownProcessIds at:index put:aProcess id.
wasBlocked ifFalse:[OperatingSystem unblockInterrupts].
^ self
].
index := index + 1
].
KnownProcessIds grow:index.
KnownProcessIds at:index put:aProcess id.
oldSize := KnownProcesses size.
(index > oldSize) ifTrue:[
newShadow := WeakArray new:(oldSize * 2).
newShadow watcher:self class.
newShadow replaceFrom:1 with:KnownProcesses.
KnownProcesses := newShadow
].
KnownProcesses at:index put:aProcess.
wasBlocked ifFalse:[OperatingSystem unblockInterrupts].
!
unRemember:aProcess
"forget aProcess - dispose processing will not consider this one"
|index wasBlocked|
wasBlocked := OperatingSystem blockInterrupts.
index := KnownProcesses identityIndexOf:aProcess.
index ~~ 0 ifTrue:[
KnownProcessIds at:index put:nil.
KnownProcesses at:index put:nil.
].
wasBlocked ifFalse:[OperatingSystem unblockInterrupts].
! !
!ProcessorScheduler methodsFor:'process creation'!
newProcessFor:aBlock
"create a new process executing aBlock.
Return a process (or nil if fail). The new process is not scheduled.
To start it running, it needs a Process>>resume."
|id p|
id := self class threadCreate:aBlock.
id isNil ifTrue:[
"
this may happen, if the VM does not support more processes,
or if it ran out of memory, when allocating internal data
structures
"
self error:'cannot create new Process'.
^ nil
].
p := Process new.
p setId:id.
p startBlock:aBlock.
p state:#light. "meaning: has no stack yet"
p setPriority:currentPriority.
"
give it a user-friendly name
"
activeProcess name notNil ifTrue:[
p name:(activeProcess name , ' (sub)')
].
self remember:p.
^ p
! !
!ProcessorScheduler methodsFor:'scheduling'!
reschedule
"switch to the highest prio runnable process
The scheduler itself is always runnable, so there is always a switch.
(if you want to implement your own scheduler stuff, uncomment below)"
^ self threadSwitch:scheduler
"/ |l p maxPri "{ Class: SmallInteger }"|
"/
"/ maxPri := SchedulingPriority.
"/ maxPri to:1 by:-1 do:[:prio |
"/ l := quiescentProcessLists at:prio.
"/ l notNil ifTrue:[
"/ p := l first.
"/ p notNil ifTrue:[
"/ activeProcess state == #active ifTrue:[
"/ activeProcess state:#run.
"/ ].
"/ ^ self threadSwitch:p
"/ ].
"/ quiescentProcessLists at:prio put:nil
"/ ]
"/ ].
"/ "
"/ no process to run - this 'cannot' happen
"/ (well, not quite: it may happen if the scheduler process is
"/ suspended - which btw. should be avoided, since noone is there
"/ to schedule processes then)
"/ "
"/
"/ MiniDebugger enterWithMessage:'fatal dispatcher should never be suspended'.
"/
"/ "try to repair by just resuming ..."
"/ activeProcess resume
!
yield
"move the currently running process to the end of the currentList
and reschedule to the first in the list, thus switching to the
next same-prio-process."
|l wasBlocked|
wasBlocked := OperatingSystem blockInterrupts.
l := quiescentProcessLists at:currentPriority.
"
debugging consistency checks - will be removed later
"
l isNil ifTrue:[
wasBlocked ifFalse:[OperatingSystem unblockInterrupts].
'oops - nil runnable list' errorPrintNL.
^ self
].
l isEmpty ifTrue:[
wasBlocked ifFalse:[OperatingSystem unblockInterrupts].
'oops - empty runnable list' errorPrintNL.
^ self
].
l size == 1 ifTrue:[
"
the running one is the only one
"
wasBlocked ifFalse:[OperatingSystem unblockInterrupts].
^ self
].
"
bring running process to the end
"
l removeFirst.
l addLast:activeProcess.
"/ wasBlocked ifFalse:[OperatingSystem unblockInterrupts].
"
and switch to first in the list
"
"/ activeProcess state:#run.
self threadSwitch:(l first).
wasBlocked ifFalse:[OperatingSystem unblockInterrupts].
!
suspend:aProcess
"remove the argument, aProcess from the list of runnable processes.
If the process is the current one, reschedule."
|pri l p s wasBlocked|
"
some debugging stuff
"
aProcess isNil ifTrue:[
MiniDebugger enterWithMessage:'nil suspend'.
^ self
].
aProcess id isNil ifTrue:[
MiniDebugger enterWithMessage:'bad suspend: already dead'.
^ self
].
aProcess == scheduler ifTrue:[
MiniDebugger enterWithMessage:'scheduler should never be suspended'.
^ self
].
wasBlocked := OperatingSystem blockInterrupts.
pri := aProcess priority.
l := quiescentProcessLists at:pri.
"
debugging consisteny checks - will be removed later
"
l isNil ifTrue:[
wasBlocked ifFalse:[OperatingSystem unblockInterrupts].
'bad suspend: empty run list' printNL.
"/ MiniDebugger enterWithMessage:'bad suspend: empty run list'.
self reschedule.
^ self
].
l remove:aProcess ifAbsent:[
wasBlocked ifFalse:[OperatingSystem unblockInterrupts].
MiniDebugger enterWithMessage:'bad suspend: not on run list'.
^ self
].
l isEmpty ifTrue:[
quiescentProcessLists at:pri put:nil.
l := nil
].
wasBlocked ifFalse:[OperatingSystem unblockInterrupts].
"
this is a bit of a kludge: allow someone else to
set the state to something like ioWait etc.
In this case, do not set to suspend.
All of this to enhance the output of the process monitor ...
"
s := aProcess state.
((s == #active) or:[s == #run]) ifTrue:[
aProcess state:#suspended.
].
(aProcess == activeProcess) ifTrue:[
"we can immediately switch sometimes"
l notNil ifTrue:[
p := l first
] ifFalse:[
p := scheduler
].
self threadSwitch:p
"/ self reschedule
].
!
resume:aProcess
"set aProcess runnable -
if its prio is higher than the currently running prio, switch to it."
|l pri wasBlocked|
aProcess == activeProcess ifTrue:[^ self].
aProcess isNil ifTrue:[^ self].
"ignore, if process is already dead"
aProcess id isNil ifTrue:[^ self].
wasBlocked := OperatingSystem blockInterrupts.
pri := aProcess priority.
l := quiescentProcessLists at:pri.
l isNil ifTrue:[
l := LinkedList new.
quiescentProcessLists at:pri put:l
] ifFalse:[
"if already running, ignore"
(l includes:aProcess) ifTrue:[
wasBlocked ifFalse:[OperatingSystem unblockInterrupts].
^ self
]
].
l addLast:aProcess.
wasBlocked ifFalse:[OperatingSystem unblockInterrupts].
(pri > currentPriority) ifTrue:[
"
its prio is higher; immediately transfer control to it
"
self threadSwitch:aProcess
] ifFalse:[
"
its prio is lower; it will have to wait for a while ...
"
aProcess state:#run
]
!
resumeForSingleSend:aProcess
"like resume, but let the process execute a single send only.
This is used by the debugger for single stepping."
|l pri wasBlocked|
aProcess == activeProcess ifTrue:[^ self].
aProcess isNil ifTrue:[^ self].
"ignore, if process is already dead"
aProcess id isNil ifTrue:[^ self].
wasBlocked := OperatingSystem blockInterrupts.
pri := aProcess priority.
l := quiescentProcessLists at:pri.
l isNil ifTrue:[
l := LinkedList new.
quiescentProcessLists at:pri put:l
] ifFalse:[
"if already running, ignore"
(l includes:aProcess) ifTrue:[
wasBlocked ifFalse:[OperatingSystem unblockInterrupts].
^ self
]
].
l addLast:aProcess.
wasBlocked ifFalse:[OperatingSystem unblockInterrupts].
(pri > currentPriority) ifTrue:[
"
its prio is higher; immediately transfer control to it
"
"/ activeProcess state:#run.
self threadSwitch:aProcess
] ifFalse:[
"
its prio is lower; it will have to wait for a while ...
"
aProcess state:#suspended
]
!
terminate:aProcess
"terminate aProcess. If its not the current process, its simply
removed from its list and destroyed. Otherwise, a switch is forced
and the process is destroyed by the next running process."
|pri id l wasBlocked|
aProcess isNil ifTrue:[^ self].
id := aProcess id.
id isNil ifTrue:[^ self]. "already dead"
aProcess setId:nil.
aProcess startBlock:nil.
wasBlocked := OperatingSystem blockInterrupts.
"remove the process from the runnable list"
pri := aProcess priority.
l := quiescentProcessLists at:pri.
(l notNil and:[l includes:aProcess]) ifTrue:[
l remove:aProcess.
l isEmpty ifTrue:[quiescentProcessLists at:pri put:nil].
].
wasBlocked ifFalse:[OperatingSystem unblockInterrupts].
aProcess exitAction notNil ifTrue:[
aProcess exitAction value.
aProcess exitAction:nil
].
aProcess state:#dead.
aProcess == activeProcess ifTrue:[
"
hard case - its the currently running process
we must have the next active process destroy this one
(we cannot destroy the chair we are sitting on ... :-)
"
zombie := id.
self unRemember:aProcess.
self threadSwitch:scheduler.
"/ self reschedule.
^ self
].
self class threadDestroy:id.
self unRemember:aProcess.
^ self
!
processTermination
"current process finished its startup block without termination,
lay him to rest now."
self terminate:activeProcess.
self threadSwitch:scheduler
"/ self reschedule
!
terminateActive
"terminate the current process
(i.e. the currently running process kills itself)"
self terminate:activeProcess
!
interruptActive
"interrupt the current process (i.e. myself)"
activeProcess interrupt
!
changePriority:prio for:aProcess
"change the priority of aProcess"
|oldList newList oldPrio newPrio wasBlocked|
oldPrio := aProcess priority.
oldPrio == prio ifTrue:[^ self].
"
check for valid argument
"
newPrio := prio.
newPrio < 1 ifTrue:[
newPrio := 1.
] ifFalse:[
newPrio >= SchedulingPriority ifTrue:[
newPrio := SchedulingPriority - 1
]
].
wasBlocked := OperatingSystem blockInterrupts.
aProcess setPriority:newPrio.
oldList := quiescentProcessLists at:oldPrio.
(oldList isNil or:[(oldList includes:aProcess) not]) ifTrue:[
wasBlocked ifFalse:[OperatingSystem unblockInterrupts].
^ self
].
oldList remove:aProcess.
oldList isEmpty ifTrue:[quiescentProcessLists at:oldPrio put:nil].
newList := quiescentProcessLists at:newPrio.
newList isNil ifTrue:[
newList := LinkedList new.
quiescentProcessLists at:newPrio put:newList
].
newList addLast:aProcess.
"if its the current process lowering its prio
or another one raising, we have to reschedule"
aProcess == activeProcess ifTrue:[
currentPriority := newPrio.
newPrio < oldPrio ifTrue:[
self threadSwitch:scheduler.
"/ self reschedule.
]
] ifFalse:[
newPrio > currentPriority ifTrue:[
"/ activeProcess state:#run.
self threadSwitch:aProcess.
]
].
wasBlocked ifFalse:[OperatingSystem unblockInterrupts].
! !
!ProcessorScheduler methodsFor:'accessing'!
currentPriority
"return the priority of the currently running process"
^ currentPriority
"Processor currentPriority"
!
activePriority
"return the priority of the currently running process.
GNU-ST compatibility; this is the same as currentPriority"
^ currentPriority
!
activeProcess
"return the currently running process"
^ activeProcess
"Processor activeProcess"
! !
!ProcessorScheduler methodsFor:'queries'!
highestPriorityRunnableProcess
"return the highest prio runnable process"
|l p maxPri "{ Class: SmallInteger }" |
maxPri := self highestPriority.
maxPri to:1 by:-1 do:[:prio |
l := quiescentProcessLists at:prio.
l notNil ifTrue:[
l isEmpty ifTrue:[
"
on the fly clear out empty lists
"
quiescentProcessLists at:prio put:nil
] ifFalse:[
p := l first.
"
if it got corrupted somehow
"
p id isNil ifTrue:[
'process with nil id removed' printNL.
l removeFirst.
^ nil.
].
^ p
].
]
].
^ nil
! !
!ProcessorScheduler methodsFor:'dispatching'!
dispatchLoop
"central dispatch loop; the scheduler process is always staying in
this method, looping forever."
dispatching == true ifTrue:[^ self].
dispatching := true.
"I made this an extra call to dispatch; this allows recompilation
of the dispatch-handling code in the running system.
"
[true] whileTrue:[
self dispatch
]
!
dispatch
"It handles timeouts and switches to the highest prio runnable process"
|any millis pri p nActions "{ Class: SmallInteger }" |
"to avoid confusion if entered twice"
"
handle all timeout actions
"
anyTimeouts ifTrue:[
self evaluateTimeouts
].
"first do a quick check using checkActions - this is needed for
devices like X-connection, where some events might be in the event
queue, so a select does not always help
"
any := false.
nActions := readChecks size.
1 to:nActions do:[:index |
|checkBlock sema action|
checkBlock := readChecks at:index.
(checkBlock notNil and:[checkBlock value]) ifTrue:[
sema := readSemaphores at:index.
sema notNil ifTrue:[
sema signalOnce.
].
any := true.
]
].
"now, someone might be runnable:"
p := self highestPriorityRunnableProcess.
p isNil ifTrue:[
"no one runnable, hard wait for event or timeout"
self waitForEventOrTimeout.
^ self
].
pri := p priority.
"want to give control to another process p.
If the switched-to processes priority is lower than the
userSchedulingPriority, we have to make certain, that the
next input or timer will bring us back for a reschedule.
This is done by enabling ioInterrupts for all file descriptors.
If ioInterrupts are not available, we schedule a timer interrupt
to interrupt us after 1/20s of a second - effectively polling
the filedescriptors. - which is very bad, since low prio processes
will be hurt in performance
- dont let benchmarks run with low prio ...
Higher prio processes must suspend, same prio ones must yield to
get back control"
"
uncommenting this will make timeouts interrupt the current process
(i.e. as if the interrupt runs at TimingPrio);
if left commented, they are handled at UserSchedulingPrio.
this will all change, when timeouts are removed and all is process driven
"
"
pri < TimingPriority ifTrue:[
anyTimeouts ifTrue:[
millis := self timeToNextTimeout.
millis == 0 ifTrue:[^ self].
]
].
"
"
if the process to run has a lower than UserInterruptPriority,
arrange for it to be interruptable by I/O.
This is done by enabling I/O-signals (if the OS supports them)
or by installing a poll-interrupt after 50ms (if the OS does not).
"
pri < UserInterruptPriority ifTrue:[
"comment out this if above is uncommented"
anyTimeouts ifTrue:[
millis := self timeToNextTimeout.
millis == 0 ifTrue:[^ self].
].
"---"
OperatingSystem supportsIOInterrupts ifTrue:[
readFds do:[:fd |
fd notNil ifTrue:[
OperatingSystem enableIOInterruptsOn:fd
].
].
] ifFalse:[
millis notNil ifTrue:[
millis := millis min:50
] ifFalse:[
millis := 50
]
]
].
millis notNil ifTrue:[
"schedule a clock interrupt after millis milliseconds"
OperatingSystem enableTimer:millis rounded.
].
"now let the process run - will come back here by reschedule
from ioInterrupt or timerInterrupt ... (running at max+1)"
"/ activeProcess state:#run.
self threadSwitch:p.
"... when we arrive here, we are back on stage"
millis notNil ifTrue:[
OperatingSystem disableTimer.
self checkForInputWithTimeout:0.
]
! !
!ProcessorScheduler methodsFor:'waiting'!
ioInterrupt
"data arrived while waiting - reschedule to bring dispatcher into play"
self threadSwitch:scheduler
"/ self reschedule
!
timerInterrupt
"timer expired while waiting - reschedule to bring dispatcher into play"
self threadSwitch:scheduler
"/ self reschedule
!
timeToNextTimeout
"return the delta-T (in millis) to next timeout, or nil if
there is none"
|aTime now minDelta n "{ Class: SmallInteger }"|
"find next timeout. since there are usually not many, just search.
If there where many, the list should be kept sorted ... keeping deltas
to next (as in Unix kernel)"
n := timeouts size.
1 to:n do:[:index |
aTime := timeouts at:index.
aTime notNil ifTrue:[
minDelta isNil ifTrue:[
now := OperatingSystem getMillisecondTime.
(OperatingSystem millisecondTime:aTime isAfter:now) ifFalse:[^ 0].
minDelta := OperatingSystem millisecondTimeDeltaBetween:aTime and:now.
] ifFalse:[
(OperatingSystem millisecondTime:aTime isAfter:now) ifFalse:[^ 0].
minDelta := minDelta min:(OperatingSystem millisecondTimeDeltaBetween:aTime and:now)
]
]
].
^ minDelta
!
waitForEventOrTimeout
"entered when no process is runnable - wait for either input on
any file descriptors to arrive or a timeout to happen.
If it makes sense, do some background garbage collection.
The idle actions are a leftover from previous ST/X releases and will
vanish (installing a low-prio process has the same effect)."
|millis limit doingGC|
doingGC := true.
[doingGC] whileTrue:[
anyTimeouts ifTrue:[
millis := self timeToNextTimeout.
(millis notNil and:[millis <= 0]) ifTrue:[
^ self "oops - hurry up checking"
].
].
"if its worth doing, collect a bit of garbage"
limit := ObjectMemory incrementalGCLimit.
doingGC := limit notNil and:[ObjectMemory oldSpaceAllocatedSinceLastGC > limit].
doingGC ifTrue:[
ObjectMemory gcStep.
].
"then do idle actions"
(idleActions size ~~ 0) ifTrue:[
idleActions do:[:aBlock |
aBlock value.
].
^ self "go back checking"
].
doingGC ifTrue:[
(self checkForInputWithTimeout:0) ifTrue:[
^ self "go back checking"
]
]
].
(self checkForInputWithTimeout:0) ifTrue:[
^ self "go back checking"
].
"no, really nothing to do - simply wait"
OperatingSystem supportsSelect ifFalse:[
"SCO instant ShitStation has a bug here,
waiting always 1 sec in the select - therefore we delay a bit and
return - effectively polling in 50ms cycles
"
OperatingSystem millisecondDelay:50.
^ self
].
millis isNil ifTrue:[
millis := 9999.
] ifFalse:[
millis := millis rounded
].
self checkForInputWithTimeout:millis
!
checkForInputWithTimeout:millis
"this is called, when there is absolutely nothing to do;
hard wait for either input to arrive or a timeout to occur."
|fd index sema action|
fd := OperatingSystem
selectOnAnyReadable:readFds
writable:writeFds
exception:nil
withTimeOut:millis.
fd notNil ifTrue:[
index := readFds indexOf:fd.
index ~~ 0 ifTrue:[
sema := readSemaphores at:index.
sema notNil ifTrue:[
sema signalOnce.
^ true
] ifFalse:[
action := readChecks at:index.
action notNil ifTrue:[
action value.
^ true
]
]
]
].
^ false
!
evaluateTimeouts
"walk through timeouts and evaluate blocks or signal semas that need to be .."
|sema now aTime block blocksToEvaluate
processes n "{ Class: SmallInteger }"|
anyTimeouts ifFalse:[ ^ self].
"have to collect the blocks first, then evaluate them. This avoids
problems due to newly inserted blocks."
now := OperatingSystem getMillisecondTime.
blocksToEvaluate := nil.
n := timeouts size.
anyTimeouts := false.
1 to:n do:[:index |
aTime := timeouts at:index.
aTime notNil ifTrue:[
(OperatingSystem millisecondTime:aTime isAfter:now) ifFalse:[
"this one should be triggered"
sema := timeoutSemaphores at:index.
sema notNil ifTrue:[
sema signalOnce.
timeoutSemaphores at:index put:nil
] ifFalse:[
"to support pure-events"
block := timeoutActions at:index.
block notNil ifTrue:[
blocksToEvaluate isNil ifTrue:[
blocksToEvaluate := OrderedCollection new:10.
processes := OrderedCollection new:10.
].
blocksToEvaluate add:block.
processes add:(timeoutProcesses at:index).
timeoutActions at:index put:nil.
timeoutProcesses at:index put:nil.
]
].
timeouts at:index put:nil.
] ifTrue:[
anyTimeouts := true
]
]
].
blocksToEvaluate notNil ifTrue:[
1 to:blocksToEvaluate size do:[:index |
PureEventDriven ifTrue:[
(blocksToEvaluate at:index) value
] ifFalse:[
(processes at:index) interruptWith:(blocksToEvaluate at:index)
]
]
]
! !
!ProcessorScheduler methodsFor:'semaphore signalling'!
signal:aSemaphore onInput:aFileDescriptor
"arrange for a semaphore to be triggered when input on aFileDescriptor
arrives."
self signal:aSemaphore onInput:aFileDescriptor orCheck:nil
!
signal:aSemaphore onInput:aFileDescriptor orCheck:aBlock
"arrange for a semaphore to be triggered when input on aFileDescriptor
arrives OR checkblock evaluates to true.
(checkBlock is used for buffered input, where a select may not detect
data already read into a buffer - as in Xlib)"
|idx wasBlocked|
wasBlocked := OperatingSystem blockInterrupts.
(readFds includes:aFileDescriptor) ifFalse:[
idx := readFds indexOf:nil.
idx ~~ 0 ifTrue:[
readFds at:idx put:aFileDescriptor.
readSemaphores at:idx put:aSemaphore.
readChecks at:idx put:aBlock
] ifFalse:[
readFds := readFds copyWith:aFileDescriptor.
readSemaphores := readSemaphores copyWith:aSemaphore.
readChecks := readChecks copyWith:aBlock.
]
].
wasBlocked ifFalse:[OperatingSystem unblockInterrupts].
!
signal:aSemaphore onOutput:aFileDescriptor
"arrange for a semaphore to be triggered when output on aFileDescriptor
is possible. (i.e. can be written without blocking)"
|idx wasBlocked|
wasBlocked := OperatingSystem blockInterrupts.
(writeFds includes:aFileDescriptor) ifFalse:[
idx := writeFds indexOf:nil.
idx ~~ 0 ifTrue:[
writeFds at:idx put:aFileDescriptor.
writeSemaphores at:idx put:aSemaphore.
] ifFalse:[
writeFds := writeFds copyWith:aFileDescriptor.
writeSemaphores := writeSemaphores copyWith:aSemaphore.
]
].
wasBlocked ifFalse:[OperatingSystem unblockInterrupts].
!
signal:aSemaphore afterSeconds:seconds
"arrange for a semaphore to be triggered after some seconds"
self signal:aSemaphore afterMilliseconds:(seconds * 1000)
!
signal:aSemaphore afterMilliseconds:millis
"arrange for a semaphore to be triggered after some milliseconds"
|now then wasBlocked|
wasBlocked := OperatingSystem blockInterrupts.
now := OperatingSystem getMillisecondTime.
then := OperatingSystem millisecondTimeAdd:now and:millis rounded.
self signal:aSemaphore atMilliseconds:then.
wasBlocked ifFalse:[OperatingSystem unblockInterrupts].
!
signal:aSemaphore atMilliseconds:aMillisecondTime
"arrange for a semaphore to be triggered at a specific millisecond time"
|index wasBlocked|
wasBlocked := OperatingSystem blockInterrupts.
index := timeoutSemaphores identityIndexOf:aSemaphore.
index ~~ 0 ifTrue:[
timeouts at:index put:aMillisecondTime
] ifFalse:[
index := timeouts indexOf:nil.
index ~~ 0 ifTrue:[
timeoutSemaphores at:index put:aSemaphore.
timeouts at:index put:aMillisecondTime.
timeoutActions at:index put:nil.
timeoutProcesses at:index put:nil
] ifFalse:[
timeoutSemaphores := timeoutSemaphores copyWith:aSemaphore.
timeouts := timeouts copyWith:aMillisecondTime.
timeoutActions := timeoutActions copyWith:nil.
timeoutProcesses := timeoutProcesses copyWith:nil
].
].
anyTimeouts := true.
wasBlocked ifFalse:[OperatingSystem unblockInterrupts].
!
disableSemaphore:aSemaphore
"disable triggering of a semaphore"
|idx wasBlocked|
wasBlocked := OperatingSystem blockInterrupts.
idx := readSemaphores identityIndexOf:aSemaphore.
idx ~~ 0 ifTrue:[
readFds at:idx put:nil.
readSemaphores at:idx put:nil.
readChecks at:idx put:nil
].
idx := timeoutSemaphores identityIndexOf:aSemaphore.
idx ~~ 0 ifTrue:[
timeouts at:idx put:nil.
timeoutSemaphores at:idx put:nil.
timeoutActions at:idx put:nil.
timeoutProcesses at:idx put:nil.
].
wasBlocked ifFalse:[OperatingSystem unblockInterrupts].
! !
!ProcessorScheduler methodsFor:'background processing'!
addIdleBlock:aBlock
"add the argument, aBlock to the list of idle-actions.
Idle blocks are evaluated whenever no other process is runnable,
and no events are pending.
Use of idle blocks is not recommended, use a low priority processes
instead, which has the same effect. They have been implemented to support
background actions in pure-event systems, where no processes are
available.
Support for idle-blocks may vanish."
|wasBlocked|
wasBlocked := OperatingSystem blockInterrupts.
idleActions isNil ifTrue:[
idleActions := OrderedCollection new
].
idleActions add:aBlock.
wasBlocked ifFalse:[OperatingSystem unblockInterrupts].
!
removeIdleBlock:aBlock
"remove the argument, aBlock from the list of idle-blocks.
Support for idle-blocks may vanish - use low prio processes instead."
|wasBlocked|
wasBlocked := OperatingSystem blockInterrupts.
idleActions notNil ifTrue:[
idleActions remove:aBlock
].
wasBlocked ifFalse:[OperatingSystem unblockInterrupts].
! !
!ProcessorScheduler methodsFor:'I/O event actions'!
enableIOAction:aBlock on:aFileDescriptor
"half-obsolete event support: arrange for aBlock to be
evaluated when input on aFileDescriptor arrives.
This is a leftover support for pure-event systems and may vanish."
|idx wasBlocked|
wasBlocked := OperatingSystem blockInterrupts.
(readFds includes:aFileDescriptor) ifFalse:[
idx := readFds indexOf:nil.
idx ~~ 0 ifTrue:[
readFds at:idx put:aFileDescriptor.
readChecks at:idx put:aBlock.
readSemaphores at:idx put:nil
] ifFalse:[
readFds := readFds copyWith:aFileDescriptor.
readChecks := readChecks copyWith:aBlock.
readSemaphores := readSemaphores copyWith:nil.
]
].
wasBlocked ifFalse:[OperatingSystem unblockInterrupts].
!
disableFd:aFileDescriptor
"disable block events on aFileDescriptor.
This is a leftover support for pure-event systems and may vanish."
|idx wasBlocked|
wasBlocked := OperatingSystem blockInterrupts.
idx := readFds indexOf:aFileDescriptor.
idx ~~ 0 ifTrue:[
readFds at:idx put:nil.
readChecks at:idx put:nil.
readSemaphores at:idx put:nil
].
wasBlocked ifFalse:[OperatingSystem unblockInterrupts].
! !
!ProcessorScheduler methodsFor:'timed block'!
addTimedBlock:aBlock afterSeconds:delta
"add the argument, aBlock to the list of time-scheduled-blocks.
to be evaluated after delta seconds. The process which installs this timed
block will be interrupted for execution of the block.
(if it is running, the interrupt will occur in whatever method it is
executing; if it is suspended, it will be resumed for the execution).
The block will be removed from the timed-block list after evaluation
(i.e. it will trigger only once)."
self addTimedBlock:aBlock for:activeProcess afterMilliseconds:(delta * 1000) rounded
!
addTimedBlock:aBlock for:aProcess afterSeconds:delta
"add the argument, aBlock to the list of time-scheduled-blocks.
to be evaluated after delta seconds. aProcess will be interrupted for
execution of the block.
(if it is running, the interrupt will occur in whatever method it is
executing; if it is suspended, it will be resumed for the execution).
The block will be removed from the timed-block list after evaluation
(i.e. it will trigger only once)."
self addTimedBlock:aBlock for:aProcess afterMilliseconds:(delta * 1000) rounded
!
addTimedBlock:aBlock afterMilliseconds:delta
"add the argument, aBlock to the list of time-scheduled-blocks; to be
evaluated after delta milliseconds. The process which installs this timed
block will be interrupted for execution of the block.
(if it is running, the interrupt will occur in whatever method it is
executing; if it is suspended, it will be resumed for the execution).
The block will be removed from the timed-block list after evaluation
(i.e. it will trigger only once)."
^ self addTimedBlock:aBlock for:activeProcess afterMilliseconds:delta
!
addTimedBlock:aBlock for:aProcess afterMilliseconds:delta
"add the argument, aBlock to the list of time-scheduled-blocks; to be
evaluated after delta milliseconds. aProcess will be interrupted for
execution of the block.
(if it is running, the interrupt will occur in whatever method it is
executing; if it is suspended, it will be resumed for the execution).
The block will be removed from the timed-block list after evaluation
(i.e. it will trigger only once)."
|now then wasBlocked|
wasBlocked := OperatingSystem blockInterrupts.
now := OperatingSystem getMillisecondTime.
then := OperatingSystem millisecondTimeAdd:now and:delta.
self addTimedBlock:aBlock for:aProcess atMilliseconds:then.
wasBlocked ifFalse:[OperatingSystem unblockInterrupts].
!
addTimedBlock:aBlock atMilliseconds:aMillisecondTime
"add the argument, aBlock to the list of time-scheduled-blocks; to be
evaluated when the millisecondClock value passes aMillisecondTime.
The process which installs this timed block will be interrupted for
execution of the block.
(if it is running, the interrupt will occur in whatever method it is
executing; if it is suspended, it will be resumed for the execution).
The block will be removed from the timed-block list after evaluation
(i.e. it will trigger only once)."
self addTimedBlock:aBlock for:activeProcess atMilliseconds:aMillisecondTime
!
addTimedBlock:aBlock for:aProcess atMilliseconds:aMillisecondTime
"add the argument, aBlock to the list of time-scheduled-blocks; to be
evaluated by aProcess when the millisecondClock value passes
aMillisecondTime.
aProcess will be interrupted for execution of the block.
(if it is running, the interrupt will occur in whatever method it is
executing; if it is suspended, it will be resumed for the execution).
The block will be removed from the timed-block list after evaluation
(i.e. it will trigger only once)."
|index wasBlocked|
wasBlocked := OperatingSystem blockInterrupts.
index := timeoutActions identityIndexOf:aBlock.
index ~~ 0 ifTrue:[
timeouts at:index put:aMillisecondTime
] ifFalse:[
index := timeouts indexOf:nil.
index ~~ 0 ifTrue:[
timeouts at:index put:aMillisecondTime.
timeoutActions at:index put:aBlock.
timeoutSemaphores at:index put:nil.
timeoutProcesses at:index put:aProcess
] ifFalse:[
timeouts := timeouts copyWith:aMillisecondTime.
timeoutActions := timeoutActions copyWith:aBlock.
timeoutSemaphores := timeoutSemaphores copyWith:nil.
timeoutProcesses := timeoutProcesses copyWith:aProcess.
].
].
anyTimeouts := true.
wasBlocked ifFalse:[OperatingSystem unblockInterrupts].
!
removeTimedBlock:aBlock
"remove the argument, aBlock from the list of time-sceduled-blocks."
|index wasBlocked|
wasBlocked := OperatingSystem blockInterrupts.
index := timeoutActions identityIndexOf:aBlock.
(index ~~ 0) ifTrue:[
timeouts at:index put:nil.
timeoutActions at:index put:nil.
timeoutSemaphores at:index put:nil.
timeoutProcesses at:index put:nil.
].
wasBlocked ifFalse:[OperatingSystem unblockInterrupts].
! !