timer: Merge the timer changes from norbert and some races fixes
Allow timers < 1 second and have the timer loop wait for the closest time. This is not completely race free yet.
This commit is contained in:
parent
699eb31206
commit
f5c5febae8
|
@ -82,3 +82,13 @@ Object extend [
|
|||
"Compat for pharo. Use it to indicate deprecated functions"
|
||||
]
|
||||
]
|
||||
|
||||
Duration extend [
|
||||
asMilliSeconds [
|
||||
^self asMilliseconds
|
||||
]
|
||||
|
||||
asDelay [
|
||||
^Delay forMilliseconds: self asMilliseconds
|
||||
]
|
||||
]
|
||||
|
|
23
Tests.st
23
Tests.st
|
@ -32,14 +32,24 @@ TestCase subclass: DispatcherTest [
|
|||
]
|
||||
]
|
||||
|
||||
TestCase subclass: TimerTest [
|
||||
TestCase subclass: TimerSchedulerTest [
|
||||
| timerScheduler |
|
||||
|
||||
<category: 'OsmoCore-Tests'>
|
||||
|
||||
tearDown [
|
||||
timerScheduler doShutDown
|
||||
]
|
||||
|
||||
setUp [
|
||||
timerScheduler := TimerScheduler new
|
||||
]
|
||||
|
||||
testTimer [
|
||||
| sem now |
|
||||
now := DateTime now.
|
||||
sem := Semaphore new.
|
||||
TimerScheduler instance scheduleInSeconds: 2 block: [
|
||||
timerScheduler scheduleInSeconds: 2 block: [
|
||||
sem signal.
|
||||
].
|
||||
|
||||
|
@ -53,14 +63,13 @@ TestCase subclass: TimerTest [
|
|||
block := [sem signal].
|
||||
|
||||
|
||||
fire1 := TimerScheduler instance scheduleInSeconds: 5 block: block.
|
||||
timer1 := TimerScheduler instance scheduleInSeconds: 3 block: block.
|
||||
timer2 := TimerScheduler instance scheduleInSeconds: 2 block: block.
|
||||
|
||||
fire1 := timerScheduler scheduleInSeconds: 5 block: block.
|
||||
timer1 := timerScheduler scheduleInSeconds: 3 block: block.
|
||||
timer2 := timerScheduler scheduleInSeconds: 2 block: block.
|
||||
timer2 cancel.
|
||||
timer1 cancel.
|
||||
|
||||
sem wait.
|
||||
self assert: sem signals = 0
|
||||
self assert: sem signals equals: 0
|
||||
]
|
||||
]
|
||||
|
|
98
Timer.st
98
Timer.st
|
@ -65,10 +65,15 @@ Object subclass: Timer [
|
|||
<category: 'management'>
|
||||
^ schedule == nil.
|
||||
]
|
||||
|
||||
remainingTime [
|
||||
<category: 'accessing'>
|
||||
^timeout - DateTime now
|
||||
]
|
||||
]
|
||||
|
||||
Object subclass: TimerScheduler [
|
||||
| queue sem loop quit processExited |
|
||||
| queue sem loop quit processExited delay |
|
||||
<category: 'OsmoCore-Timer'>
|
||||
<comment: 'I can help to fire things at the right time. Right now I
|
||||
only work on seconds granularity because Time has no direct access to
|
||||
|
@ -84,12 +89,30 @@ bit difficult to do this race free.'>
|
|||
|
||||
TimerScheduler class >> new [
|
||||
<category: 'private'>
|
||||
^ super new
|
||||
^self basicNew
|
||||
initialize;
|
||||
addToBeFinalized;
|
||||
yourself
|
||||
]
|
||||
|
||||
TimerScheduler class >> processName [
|
||||
<category: 'accessing'>
|
||||
^'Osmo Timers'
|
||||
]
|
||||
|
||||
doStartUp [
|
||||
"Nothing for GST"
|
||||
]
|
||||
|
||||
doShutDown [
|
||||
"Nothing for GST"
|
||||
]
|
||||
|
||||
dispatchTimers [
|
||||
<category: 'delay_loop'>
|
||||
OsmoDispatcher dispatchBlock: [self fireTimers: DateTime now]
|
||||
]
|
||||
|
||||
finalize [
|
||||
<category: 'private'>
|
||||
quit := true.
|
||||
|
@ -105,39 +128,70 @@ bit difficult to do this race free.'>
|
|||
|
||||
startLoop [
|
||||
<category: 'creation'>
|
||||
|
||||
processExited := Semaphore new.
|
||||
|
||||
loop := [[self runTimers] ensure: [
|
||||
processExited signal.
|
||||
loop := nil]] newProcess.
|
||||
loop name: 'Osmo Timers'.
|
||||
loop resume.
|
||||
loop := [[self runTimers] ensure:
|
||||
[processExited signal.
|
||||
loop := nil]] newProcess.
|
||||
loop name: self class processName.
|
||||
loop resume
|
||||
]
|
||||
|
||||
scheduleInSeconds: aDelay block: aBlock [
|
||||
| sched |
|
||||
signalDelay [
|
||||
"Called with sem critical being consumed"
|
||||
delay ifNotNil: [delay signal].
|
||||
]
|
||||
|
||||
scheduleIn: aDuration block: aBlock [
|
||||
<category: 'schedule'>
|
||||
sched := (Timer on: self)
|
||||
| timer currentFirst |
|
||||
timer := (Timer on: self)
|
||||
block: aBlock;
|
||||
timeout: (DateTime now + (Duration milliseconds: 1000 * aDelay));
|
||||
timeout: DateTime now + aDuration;
|
||||
yourself.
|
||||
|
||||
sem critical: [
|
||||
queue add: sched.
|
||||
].
|
||||
currentFirst := queue isEmpty ifFalse: [queue first].
|
||||
queue add: timer.
|
||||
loop isSuspended
|
||||
ifTrue: [loop resume]
|
||||
ifFalse: [
|
||||
"if the first item in the queue has changed we need to readjust the delay
|
||||
to wait for. Signalling the waiting delay will enter the recalculation of a new
|
||||
expire time"
|
||||
|
||||
^ sched
|
||||
currentFirst == queue first
|
||||
ifFalse: [self signalDelay]]].
|
||||
^timer
|
||||
]
|
||||
|
||||
scheduleInSeconds: aNumber block: aBlock [
|
||||
<category: 'schedule'>
|
||||
^self scheduleIn: (Duration fromSeconds: aNumber) block: aBlock
|
||||
]
|
||||
|
||||
runTimers [
|
||||
<category: 'delay_loop'>
|
||||
<category: 'delay_loop'>
|
||||
|
||||
[quit] whileFalse: [ | now |
|
||||
(Delay forSeconds: 1) wait.
|
||||
now := DateTime now.
|
||||
OsmoDispatcher dispatchBlock: [self fireTimers: now].
|
||||
]
|
||||
[quit] whileFalse: [
|
||||
| timer |
|
||||
sem critical: [queue isEmpty ifFalse: [timer := queue first]].
|
||||
timer isNil
|
||||
ifTrue: [
|
||||
"nothing to do. No need to poll an empty queue. Remove delay to get rid of
|
||||
a false resumptionTime. Suspend the process. The process will be resumed
|
||||
when an item is added. Please note that Processor activeProcess == loop will
|
||||
hold here."
|
||||
|
||||
delay := nil.
|
||||
loop suspend]
|
||||
ifFalse: [
|
||||
"either a timer has expired and we process it or we wait for the first item in
|
||||
the queue to expire"
|
||||
|
||||
| offset |
|
||||
(offset := timer remainingTime) asMilliSeconds > 0
|
||||
ifTrue: [(delay := offset asDelay) wait]
|
||||
ifFalse: [self dispatchTimers]]]
|
||||
]
|
||||
|
||||
fireTimers: now [
|
||||
|
|
|
@ -19,8 +19,12 @@ TimerScheduler extend [
|
|||
<category: 'PharoHacks'>
|
||||
loop ifNil: [^self].
|
||||
quit := true.
|
||||
loop isSuspended ifTrue: [loop resume].
|
||||
delay ifNotNil: [:the_delay | the_delay signalWaitingProcess].
|
||||
processExited wait.
|
||||
Transcript nextPutAll: 'Stopped the TimerScheduler process'; cr.
|
||||
Transcript
|
||||
nextPutAll: 'Stopped the TimerScheduler process';
|
||||
cr
|
||||
]
|
||||
|
||||
doStartUp [
|
||||
|
@ -30,6 +34,16 @@ TimerScheduler extend [
|
|||
quit := false.
|
||||
self startLoop.
|
||||
]
|
||||
|
||||
signalDelay [
|
||||
"Called with sem critical being consumed"
|
||||
delay ifNotNil: [delay signalWaitingProcess].
|
||||
]
|
||||
|
||||
scheduleInSeconds: aNumber block: aBlock [
|
||||
<category: 'schedule'>
|
||||
^self scheduleIn: aNumber seconds block: aBlock
|
||||
]
|
||||
]
|
||||
|
||||
Dispatcher class extend [
|
||||
|
|
|
@ -10,7 +10,7 @@
|
|||
|
||||
<test>
|
||||
<sunit>Osmo.DispatcherTest</sunit>
|
||||
<sunit>Osmo.TimerTest</sunit>
|
||||
<sunit>Osmo.TimerSchedulerTest</sunit>
|
||||
<filein>Tests.st</filein>
|
||||
<sunit>Osmo.StringFormatTest</sunit>
|
||||
<filein>ExtensionTest.st</filein>
|
||||
|
|
Reference in New Issue