@@ -19,6 +19,7 @@ import (
19
19
"encoding/hex"
20
20
"fmt"
21
21
"maps"
22
+ "slices"
22
23
"sync"
23
24
"time"
24
25
@@ -252,9 +253,7 @@ func (rt *TimerGroupRuntime) fullRefreshTimers() {
252
253
253
254
func (rt * TimerGroupRuntime ) tryTriggerTimerEvents () {
254
255
now := rt .nowFunc ()
255
- var retryTimerIDs []string
256
- var retryTimerKeys []string
257
- var busyWorkers map [string ]struct {}
256
+ var readyTimers []* timerCacheItem
258
257
rt .cache .iterTryTriggerTimers (func (timer * api.TimerRecord , tryTriggerTime time.Time , nextEventTime * time.Time ) bool {
259
258
if tryTriggerTime .After (now ) {
260
259
return false
@@ -264,9 +263,45 @@ func (rt *TimerGroupRuntime) tryTriggerTimerEvents() {
264
263
return true
265
264
}
266
265
266
+ if readyTimers == nil {
267
+ readyTimers = make ([]* timerCacheItem , 0 , 8 )
268
+ }
269
+
270
+ readyTimers = append (readyTimers , & timerCacheItem {
271
+ timer : timer ,
272
+ nextEventTime : nextEventTime ,
273
+ })
274
+ return true
275
+ })
276
+
277
+ if len (readyTimers ) == 0 {
278
+ return
279
+ }
280
+
281
+ // resort timer to make sure the timer has the smallest nextEventTime has a higher priority to trigger
282
+ slices .SortFunc (readyTimers , func (a , b * timerCacheItem ) int {
283
+ if a .nextEventTime == nil || b .nextEventTime == nil {
284
+ if a .nextEventTime != nil {
285
+ return 1
286
+ }
287
+
288
+ if b .nextEventTime != nil {
289
+ return - 1
290
+ }
291
+
292
+ return 0
293
+ }
294
+ return a .nextEventTime .Compare (* b .nextEventTime )
295
+ })
296
+
297
+ var retryTimerIDs []string
298
+ var retryTimerKeys []string
299
+ var busyWorkers map [string ]struct {}
300
+ for i , item := range readyTimers {
301
+ timer := item .timer
267
302
worker , ok := rt .ensureWorker (timer .HookClass )
268
303
if ! ok {
269
- return true
304
+ continue
270
305
}
271
306
272
307
eventID := timer .EventID
@@ -284,20 +319,22 @@ func (rt *TimerGroupRuntime) tryTriggerTimerEvents() {
284
319
285
320
select {
286
321
case <- rt .ctx .Done ():
287
- return false
322
+ return
288
323
case worker .ch <- req :
289
324
rt .cache .setTimerProcStatus (timer .ID , procTriggering , eventID )
290
325
default :
291
326
if busyWorkers == nil {
292
- busyWorkers = make (map [string ]struct {})
327
+ busySize := len (readyTimers ) - i
328
+ retryTimerIDs = make ([]string , 0 , busySize )
329
+ retryTimerKeys = make ([]string , 0 , busySize )
330
+ busyWorkers = make (map [string ]struct {}, busySize )
293
331
}
294
332
295
333
busyWorkers [timer .HookClass ] = struct {}{}
296
334
retryTimerIDs = append (retryTimerIDs , timer .ID )
297
335
retryTimerKeys = append (retryTimerKeys , fmt .Sprintf ("[%s] %s" , timer .Namespace , timer .Key ))
298
336
}
299
- return true
300
- })
337
+ }
301
338
302
339
if len (retryTimerIDs ) > 0 {
303
340
busyWorkerList := make ([]string , 0 , len (busyWorkers ))
0 commit comments