IWF-274: Optimize Timer creation #529
Conversation
|
|
||
| type sortedTimers struct { | ||
| status service.InternalTimerStatus | ||
| // Ordered slice of all timers being awaited on |
There was a problem hiding this comment.
Nit: add note for sorting on descending order
| "github.com/indeedeng/iwf/service" | ||
| ) | ||
|
|
||
| type sortedTimers struct { |
There was a problem hiding this comment.
We probably can still call it "TimerScheduler" and move tp.createGreedyTimerScheduler(ctx, continueAsNewCounter) under it.
So that this can be in a spearate file as timerScheduler.go and this processor can call NewTimerScheduler(...) to initiate it.
type TimerScheduler struct{
PendingSchedules TimerInfo
ScheduledTimes []int64
}
func NewTimerScheduler(...) *TimerScheduler{
createGreedyTimerSchedulerCoroutine(ctx, continueAsNewCounter)
}
| // start some single thread that manages timers | ||
| tp.createGreedyTimerScheduler(ctx, continueAsNewCounter) | ||
|
|
||
| err := provider.SetQueryHandler(ctx, service.GetCurrentTimerInfosQueryType, func() (service.GetCurrentTimerInfosQueryResponse, error) { |
There was a problem hiding this comment.
It would be nice to return the TimerScheduler here (it will make it easier for debug, and potentially add some advanced test cases for this greedyTimerProcessor.
But we probably need to move GetCurrentTimerInfosQueryResponse to this package in order to do that.
| } | ||
| insertIndex = i + 1 | ||
| } | ||
| t.timers = append( |
There was a problem hiding this comment.
I wonder it would be more memory-efficient to break it into two operations:
t.timers = append(t.timers[:insertIndex], toAdd)
t.timers = append(t.timers, t.timers[insertIndex:]...)
And also maybe a little more readable
There was a problem hiding this comment.
we probably need to add some check here:
func main() {
arr := []int{1, 2}
fmt.Println(arr)
fmt.Println(arr[:1])
fmt.Println(arr[:2])
fmt.Println(arr[3:])
}
will panic because slice bounds out of range
|
|
||
| func (t *sortedTimers) removeTimer(toRemove *service.TimerInfo) { | ||
| for i, timer := range t.timers { | ||
| if toRemove == timer { |
There was a problem hiding this comment.
I think you need to check the value is equal, not the ref.
if *toRemove == *timer
| for i := len(createdTimers) - 1; i >= 0; i-- { | ||
| if createdTimers[i] > now { | ||
| createdTimers = createdTimers[:i+1] | ||
| break | ||
| } | ||
| } |
There was a problem hiding this comment.
Nit: maybe also move to pruneScheduledTimes and rename the above to prunePendingSchedules
| } | ||
| } | ||
| next := t.pendingTimers.pruneToNextTimer(now) | ||
| return (next != nil && (len(createdTimers) == 0 || next.FiringUnixTimestampSeconds < createdTimers[len(createdTimers)-1])) || continueAsNewCounter.IsThresholdMet() |
There was a problem hiding this comment.
maybe add a few comments here for the cases:
- "next != nil && (len(createdTimers) == 0" means we should schedule a next timer because there are none
- "next.FiringUnixTimestampSeconds < createdTimers[len(createdTimers)-1]))“ means we should schedule a next timer because the next pending one is earlier than the earliest in scheduledTimes
- continueAsNewCounter.IsThresholdMet() continueAsNew is needed
| next := t.pendingTimers.pruneToNextTimer(now) | ||
| //next := t.pendingTimers.getEarliestTimer() | ||
| // only create a new timer when a pending timer exists before the next existing timer fires | ||
| if next != nil && (len(createdTimers) == 0 || next.FiringUnixTimestampSeconds < createdTimers[len(createdTimers)-1]) { |
There was a problem hiding this comment.
I feel like we don't have to do this check again, if it's not continueAsNew, it should be the case already?
| if next != nil && (len(createdTimers) == 0 || next.FiringUnixTimestampSeconds < createdTimers[len(createdTimers)-1]) { | ||
| fireAt := next.FiringUnixTimestampSeconds | ||
| duration := time.Duration(fireAt-now) * time.Second | ||
| t.provider.NewTimer(ctx, duration) |
There was a problem hiding this comment.
This NewTimer is a non-blocking one.
At first glance I thought it should be t.provider.Sleep(...) then I remember that we had a similar discussion when implemeting it for Wayfinder LoL.
So we don't need to wait here because we rely on the fact that when the timer fired, workflow task will get triggered and evaluate the Await condition, right?
Maybe add more comments here too.
| return service.TimerSkipped | ||
| } | ||
|
|
||
| _ = t.provider.Await(ctx, func() bool { |
There was a problem hiding this comment.
Nit: add a comment here that the timers in the TimerScheduler will trigger a workflow task when the timer fire, hence it will drive the workflow task to evalaute timer.FiringUnixTimestampSeconds <= t.provider.Now(ctx).Unix().
That was a tricky part too in our Wayfinder code :P
|
Would be great to add a new test case to it -- especially the idleTimeout pattern, which is that main use case that this optimization is for. Most of the code can be from S1-1 will wait for timer, S1-2 will wait for timer+signal. We can change it to Both S1-1 and S1-2 will wait for signal+timer and only the second timer will fired. We will expect only one timer was started in Temporal. |
Description
Checklist
Related Issue
Closes #issue_number