nextSequenceGreaterThan increments _sync:seq such that it's greater than existingSequence + s.sequenceBatchSize In the case where our local s.max < _sync:seq (another node has incremented _sync:seq), we may be releasing sequences greater than existingSequence, but we will only ever release sequences
(ctx context.Context, existingSequence uint64)
| 204 | // In the case where our local s.max < _sync:seq (another node has incremented _sync:seq), we may be releasing |
| 205 | // sequences greater than existingSequence, but we will only ever release sequences allocated by this node's incr operation |
| 206 | func (s *sequenceAllocator) nextSequenceGreaterThan(ctx context.Context, existingSequence uint64) (sequence uint64, releasedSequenceCount uint64, err error) { |
| 207 | |
| 208 | targetSequence := existingSequence + 1 |
| 209 | s.mutex.Lock() |
| 210 | // If the target sequence is less than or equal to one we've already allocated, can assign the sequence in the standard way |
| 211 | if targetSequence <= s.last { |
| 212 | sequence, sequencesReserved, err := s._nextSequence(ctx) |
| 213 | s.mutex.Unlock() |
| 214 | if err != nil { |
| 215 | return 0, 0, err |
| 216 | } |
| 217 | if sequencesReserved { |
| 218 | s.reserveNotify <- struct{}{} |
| 219 | } |
| 220 | return sequence, 0, nil |
| 221 | } |
| 222 | |
| 223 | // If the target sequence is in our existing batch (between s.last and s.max), we want to release all unused sequences in the batch earlier |
| 224 | // than targetSequence, and then assign as targetSequence |
| 225 | if targetSequence <= s.max { |
| 226 | releaseFrom := s.last + 1 |
| 227 | s.last = targetSequence |
| 228 | s.dbStats.LastSequenceAssignedValue.Set(int64(targetSequence)) |
| 229 | s.mutex.Unlock() |
| 230 | if releaseFrom < targetSequence { |
| 231 | released, err := s.releaseSequenceRange(ctx, releaseFrom, targetSequence-1) |
| 232 | if err != nil { |
| 233 | base.WarnfCtx(ctx, "Error returned when releasing sequence range [%d-%d] from existing batch. Will be handled by skipped sequence handling. Error:%v", releaseFrom, targetSequence-1, err) |
| 234 | } |
| 235 | releasedSequenceCount += released |
| 236 | } |
| 237 | s.dbStats.SequenceAssignedCount.Add(1) |
| 238 | return targetSequence, releasedSequenceCount, nil |
| 239 | |
| 240 | } |
| 241 | |
| 242 | // At this point we need to allocate a sequence that's larger than what's in our current batch, so we first need to release the current batch. |
| 243 | var numReleasedBatch uint64 |
| 244 | numReleasedBatch, err = s._releaseCurrentBatch(ctx) |
| 245 | if err != nil { |
| 246 | base.InfofCtx(ctx, base.KeyCache, "Unable to release current batch during nextSequenceGreaterThan for existing sequence %d. Will be handled by skipped sequence handling. %v", existingSequence, err) |
| 247 | } |
| 248 | releasedSequenceCount += numReleasedBatch |
| 249 | |
| 250 | syncSeq, err := s.getSequence() |
| 251 | if err != nil { |
| 252 | base.WarnfCtx(ctx, "Unable to fetch current sequence during nextSequenceGreaterThan for existing sequence %d. Error:%v", existingSequence, err) |
| 253 | s.mutex.Unlock() |
| 254 | return 0, 0, err |
| 255 | } |
| 256 | |
| 257 | // check for rollback of _sync:seq before continuing |
| 258 | if syncSeq < s.last { |
| 259 | // rollback of _sync:seq detected |
| 260 | syncSeq, err = s._fixSyncSeqRollback(ctx, syncSeq, s.last) |
| 261 | if err != nil { |
| 262 | s.mutex.Unlock() |
| 263 | return 0, 0, err |