forked from gocraft/work
-
Notifications
You must be signed in to change notification settings - Fork 0
/
periodic_enqueuer.go
154 lines (132 loc) · 3.45 KB
/
periodic_enqueuer.go
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
package work
import (
"context"
"fmt"
"math/rand"
"time"
"github.com/redis/go-redis/v9"
"github.com/robfig/cron/v3"
)
const (
periodicEnqueuerSleep = 2 * time.Minute
periodicEnqueuerHorizon = 4 * time.Minute
)
type periodicEnqueuer struct {
namespace string
redisClient *redis.Client
logger Logger
periodicJobs []*periodicJob
stopChan chan struct{}
doneStoppingChan chan struct{}
}
type periodicJob struct {
jobName string
spec string
schedule cron.Schedule
}
func newPeriodicEnqueuer(namespace string, redisClient *redis.Client, periodicJobs []*periodicJob, logger Logger) *periodicEnqueuer {
return &periodicEnqueuer{
namespace: namespace,
redisClient: redisClient,
periodicJobs: periodicJobs,
logger: logger,
stopChan: make(chan struct{}),
doneStoppingChan: make(chan struct{}),
}
}
func (pe *periodicEnqueuer) start() {
go pe.loop()
}
func (pe *periodicEnqueuer) stop() {
pe.stopChan <- struct{}{}
<-pe.doneStoppingChan
}
func (pe *periodicEnqueuer) loop() {
// Begin reaping periodically
timer := time.NewTimer(periodicEnqueuerSleep + time.Duration(rand.Intn(30))*time.Second)
defer timer.Stop()
if pe.shouldEnqueue() {
err := pe.enqueue()
if err != nil {
if pe.logger != nil {
pe.logger.Printf("periodic_enqueuer.loop.enqueue: %s", err)
}
}
}
for {
select {
case <-pe.stopChan:
pe.doneStoppingChan <- struct{}{}
return
case <-timer.C:
timer.Reset(periodicEnqueuerSleep + time.Duration(rand.Intn(30))*time.Second)
if pe.shouldEnqueue() {
err := pe.enqueue()
if err != nil {
if pe.logger != nil {
pe.logger.Printf("periodic_enqueuer.loop.enqueue: %s", err)
}
}
}
}
}
}
func (pe *periodicEnqueuer) enqueue() error {
now := nowEpochSeconds()
nowTime := time.Unix(now, 0)
horizon := nowTime.Add(periodicEnqueuerHorizon)
for _, pj := range pe.periodicJobs {
for t := pj.schedule.Next(nowTime); t.Before(horizon); t = pj.schedule.Next(t) {
epoch := t.Unix()
id := makeUniquePeriodicID(pj.jobName, pj.spec, epoch)
job := &Job{
Name: pj.jobName,
ID: id,
// This is technically wrong, but this lets the bytes be identical for the same periodic job instance. If we don't do this, we'd need to use a different approach -- probably giving each periodic job its own history of the past 100 periodic jobs, and only scheduling a job if it's not in the history.
EnqueuedAt: epoch,
Args: nil,
}
rawJSON, err := job.serialize()
if err != nil {
return err
}
if err := pe.redisClient.ZAdd(
context.TODO(),
redisKeyScheduled(pe.namespace),
redis.Z{
Score: float64(epoch),
Member: rawJSON,
},
).Err(); err != nil {
return err
}
}
}
if err := pe.redisClient.Set(
context.TODO(),
redisKeyLastPeriodicEnqueue(pe.namespace),
now,
0,
).Err(); err != nil {
return err
}
return nil
}
func (pe *periodicEnqueuer) shouldEnqueue() bool {
lastEnqueue, err := pe.redisClient.Get(
context.TODO(),
redisKeyLastPeriodicEnqueue(pe.namespace),
).Int64()
if err == redis.Nil {
return true
} else if err != nil {
if pe.logger != nil {
pe.logger.Printf("periodic_enqueuer.should_enqueue: %s", err)
}
return true
}
return lastEnqueue < (nowEpochSeconds() - int64(periodicEnqueuerSleep/time.Minute))
}
func makeUniquePeriodicID(name, spec string, epoch int64) string {
return fmt.Sprintf("periodic:%s:%s:%d", name, spec, epoch)
}