LockMutex locks the value stored at k with a mutex with identifier id. It stores the lock at LockKey(k) and list at ListKey(k).
(ctx context.Context, r redis.Cmdable, k, id string, expiration time.Duration)
| 782 | // LockMutex locks the value stored at k with a mutex with identifier id. |
| 783 | // It stores the lock at LockKey(k) and list at ListKey(k). |
| 784 | func LockMutex(ctx context.Context, r redis.Cmdable, k, id string, expiration time.Duration) error { |
| 785 | defer trace.StartRegion(ctx, "lock mutex").End() |
| 786 | |
| 787 | var hasDeadline bool |
| 788 | dl, ok := ctx.Deadline() |
| 789 | if ok { |
| 790 | hasDeadline = !dl.IsZero() |
| 791 | } |
| 792 | |
| 793 | lockKey := LockKey(k) |
| 794 | listKey := ListKey(k) |
| 795 | expMS := milliseconds(expiration) |
| 796 | for { |
| 797 | ttlMS, err := lockMutexScript.Run(ctx, r, []string{lockKey, listKey}, id, expMS).Int64() |
| 798 | if err != nil { |
| 799 | return ConvertError(err) |
| 800 | } |
| 801 | if ttlMS < 0 { |
| 802 | panic(fmt.Errorf("negative TTL returned: %d ms", ttlMS)) |
| 803 | } |
| 804 | if ttlMS == 0 { |
| 805 | return nil |
| 806 | } |
| 807 | |
| 808 | timeout := time.Duration(ttlMS) * time.Millisecond |
| 809 | if hasDeadline { |
| 810 | until := time.Until(dl) |
| 811 | if until < timeout { |
| 812 | timeout = until |
| 813 | } |
| 814 | } |
| 815 | popRes, err := r.BLPop(ctx, timeout, listKey).Result() |
| 816 | if err != nil && !errors.Is(err, redis.Nil) { |
| 817 | return ConvertError(err) |
| 818 | } |
| 819 | select { |
| 820 | case <-ctx.Done(): |
| 821 | if errors.Is(err, redis.Nil) { |
| 822 | return ctx.Err() |
| 823 | } |
| 824 | // Pass the lock to next caller. |
| 825 | if err := unlockMutexScript.Run(ctx, r, []string{lockKey, listKey}, popRes[1], expMS).Err(); err != nil { |
| 826 | log.FromContext(ctx).WithError(ConvertError(err)).Error("Failed to pass mutex to next caller") |
| 827 | } |
| 828 | return ctx.Err() |
| 829 | default: |
| 830 | } |
| 831 | if errors.Is(err, redis.Nil) { |
| 832 | continue |
| 833 | } |
| 834 | |
| 835 | // Attempt to take over the lock from previous caller. |
| 836 | v, err := takeMutexLockScript.Run(ctx, r, []string{lockKey, listKey}, popRes[1], expMS, id).Int64() |
| 837 | if err != nil { |
| 838 | return ConvertError(err) |
| 839 | } |
| 840 | if v == 1 { |
| 841 | return nil |