osmo-st-core/Timer.st

"
 (C) 2011 by Holger Hans Peter Freyther
 All Rights Reserved

 This program is free software: you can redistribute it and/or modify
 it under the terms of the GNU Affero General Public License as
 published by the Free Software Foundation, either version 3 of the
 License, or (at your option) any later version.

 This program is distributed in the hope that it will be useful,
 but WITHOUT ANY WARRANTY; without even the implied warranty of
 MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 GNU Affero General Public License for more details.

 You should have received a copy of the GNU Affero General Public License
 along with this program.  If not, see <http://www.gnu.org/licenses/>.
"

Object subclass: Timer [
    | schedule timeout block |

    <category: 'OsmoCore-Timer'>
    <comment: 'This is a receipt for an active timer'>

    Timer class >> on: aSchedule [
        <category: 'creation'>
        ^ self new
            schedule: aSchedule;
            yourself
    ]

    timeout [
        <category: 'accessing'>
        ^ timeout
    ]

    schedule: aSchedule [
        <category: 'creation'>
        schedule := aSchedule.
    ]

    timeout: aTimeout [
        <category: 'creation'>
        timeout := aTimeout.
    ]

    block: aBlock [
        <category: 'creation'>
        block := aBlock
    ]

    fire [
        <category: 'execution'>
        block value
    ]

    cancel [
        <category: 'management'>
        "Remember that the timer is gone."
        schedule := nil.
        block := nil.
    ]

    isCanceled [
        <category: 'management'>
        ^ schedule == nil.
    ]

    remainingTime [
	<category: 'accessing'>
	^timeout - DateTime now
    ]
]

Object subclass: TimerScheduler [
    | queue sem loop quit processExited delay loopSem |
    <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
milliseconds. Also I run a loop every second. I should use a Semaphore to
signal the process about a change of the closest time but it might be a
bit difficult to do this race free.'>

    TimerScheduler class >> instance [
        <category: 'singleton'>
        ^ Smalltalk at: #OsmoTimeScheduler ifAbsentPut: [TimerScheduler new].
    ]


    TimerScheduler class >> new [
        <category: 'private'>
        ^self basicNew
            initialize;
            addToBeFinalized;
            yourself
    ]

    TimerScheduler class >> processName [
        <category: 'accessing'>
        ^'Osmo Timers'
    ]

    doStartUp [
        "Nothing for GST"
    ]

    doShutDown [
        "Nothing for GST"
        loop ifNil: [^self].
        quit := true.
        sem critical: [
            loopSem ifNotNil: [loopSem signal]].
        delay ifNotNil: [:the_delay | the_delay signal].
        processExited wait.
        Transcript nextPutAll: 'Stopped the TimerScheduler process'; cr
    ]

    dispatchTimers [
        <category: 'delay_loop'>
        OsmoDispatcher dispatchBlock: [self fireTimers: DateTime now]
    ]

    finalize [
        <category: 'private'>
        quit := true.
    ]

    initialize [
        <category: 'private'>
        queue := SortedCollection sortBlock: [:a :b | a timeout < b timeout].
        sem := Semaphore forMutualExclusion.
        quit := false.
        self startLoop.
    ]

    startLoop [
        <category: 'creation'>
        processExited := Semaphore new.
        loop := [[self runTimers] ensure:
                    [processExited signal.
                     loop := nil]] newProcess.
        loop name: self class processName.
        loop resume
    ]

    signalDelay [
        "Called with sem critical being consumed"
        delay ifNotNil: [delay signal].
    ]

    scheduleIn: aDuration block: aBlock [
        <category: 'schedule'>
        | timer currentFirst |
        timer := (Timer on: self)
                    block: aBlock;
                    timeout: DateTime now + aDuration;
                    yourself.

        sem critical: [
            currentFirst := queue isEmpty ifFalse: [queue first].
            queue add: timer.

            "Make sure the loopSem is waking up at least once."
            loopSem ifNotNil: [loopSem signal].
            "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"
            currentFirst == queue first
                ifFalse: [self signalDelay]].
	^timer
    ]

    scheduleInSeconds: aNumber block: aBlock [
        <category: 'schedule'>
        ^self scheduleIn: (Duration fromSeconds: aNumber) block: aBlock
    ]

    runTimers [
	<category: 'delay_loop'>

        [quit] whileFalse: [
            | timer |
            sem critical: [
                queue isEmpty ifFalse: [timer := queue first].
                loopSem := Semaphore new.
            ].

            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.
                    loopSem wait]
                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 [
        <category: 'private'>

        "Now execute the timers. One way or another this is crazy. If we have
        a long blocking application or a deadlock the timer queue will get
        stuck. But if we run this in a new process a later process might be run
        before this process, changing the order of the timers."
        "Only this process will remove items, this is why we can check isEmpty
        without having the lock"
        [queue isEmpty or: [queue first timeout > now]] whileFalse: [ | each |
            each := sem critical: [queue removeFirst].
            each isCanceled ifFalse: [
                [each fire] on: Error do: [:e |
                e logException:
                    ('Execution of timer failed: ', e messageText) area: #timer.
            ]].
        ]
    ]
]