Skip to content

Commit 0990fd1

Browse files
authored
TWEAK: pull only locks once (#104)
1 parent a6c4fd9 commit 0990fd1

File tree

1 file changed

+26
-17
lines changed

1 file changed

+26
-17
lines changed

sidekiq/queue.go

Lines changed: 26 additions & 17 deletions
Original file line numberDiff line numberDiff line change
@@ -213,24 +213,8 @@ func (q *sidekiqQueue) Pull(opt *PullOptions) error {
213213
if err != nil {
214214
return err
215215
}
216-
pull := func() error {
217-
lock := &redislock.Lock{
218-
Client: q.client,
219-
Key: fmt.Sprintf("%s:sidekiq-queue-pull:%s", opt.SidekiqNamespace, opt.SidekiqQueue),
220-
ID: uuid.NewString(),
221-
At: time.Now(),
222-
ExpireInSec: 30,
223-
MaxAcquirers: 1,
224-
}
225-
acquired, err := lock.Acquire()
226-
if err != nil {
227-
return err
228-
}
229-
if !acquired {
230-
return redis.Nil
231-
}
232-
defer lock.Release()
233216

217+
pull := func() error {
234218
res, err := q.dequeueScript.Run(context.Background(), q.client, nil,
235219
opt.SidekiqNamespace,
236220
opt.SidekiqQueue,
@@ -286,7 +270,32 @@ func (q *sidekiqQueue) Pull(opt *PullOptions) error {
286270
return nil
287271
}
288272

273+
const timeoutSec = 30
274+
ctx, cancel := context.WithTimeout(context.Background(), timeoutSec*time.Second)
275+
defer cancel()
276+
lock := &redislock.Lock{
277+
Client: q.client,
278+
Key: fmt.Sprintf("%s:sidekiq-queue-pull:%s", opt.SidekiqNamespace, opt.SidekiqQueue),
279+
ID: uuid.NewString(),
280+
At: time.Now(),
281+
ExpireInSec: timeoutSec,
282+
MaxAcquirers: 1,
283+
}
284+
acquired, err := lock.Acquire()
285+
if err != nil {
286+
return err
287+
}
288+
if !acquired {
289+
return nil
290+
}
291+
defer lock.Release()
292+
289293
for {
294+
select {
295+
case <-ctx.Done():
296+
return nil
297+
default:
298+
}
290299
err := pull()
291300
if err != nil {
292301
if errors.Is(err, redis.Nil) {

0 commit comments

Comments
 (0)