// RequeueWithoutBackoff sends a REQ command to the nsqd which
// sent this message, using the supplied delay.
//
// Notably, using this method to respond does not trigger a backoff
// event on the configured Delegate.
496237
func (m *Message) RequeueWithoutBackoff(delay time.Duration) {
	m.doRequeue(delay, false)
}
300534
func (c *Client) DoRequestWithoutRedirects(r *Request) (*http.Response, error) {
	prevCheckRedirect := c.Config.HttpClient.CheckRedirect
	c.Config.HttpClient.CheckRedirect = func(httpReq *http.Request, via []*http.Request) error {
		return http.ErrUseLastResponse
	}
	defer func() {
		c.Config.HttpClient.CheckRedirect = prevCheckRedirect
	}()
	return c.DoRequest(r)
}
230835
func (tt *timeoutThread) waitForNonEmptyQueue() bool {
	tt.mu.Lock()
	queueNotEmpty := tt.queueNotEmpty
	tt.mu.Unlock()

	select {
	// a) Always check these channels, regardless of the state.
	case <-tt.maxDuration.C:
		// Max duration is up. Stop buffering. Do not error out entries explicitly.
		tt.sb.stopBufferingDueToMaxDuration()
		return true
	case <-tt.stopChan:
		// Failover ended before timeout. Do nothing.
		return true
	// b) State-specific check.
	case <-queueNotEmpty:
		// At least one entry present. Check its timeout in the next iteration.
		return false
	}
}
190765
func (c *Channel) RequeueMessage(clientID int64, id MessageID, timeout time.Duration) error {
	// remove from inflight first
	msg, err := c.popInFlightMessage(clientID, id)
	if err != nil {
		return err
	}
	c.removeFromInFlightPQ(msg)
	atomic.AddUint64(&c.requeueCount, 1)

	if timeout == 0 {
		c.exitMutex.RLock()
		if c.Exiting() {
			c.exitMutex.RUnlock()
			return errors.New("exiting")
		}
		err := c.put(msg)
		c.exitMutex.RUnlock()
		return err
	}

	// deferred requeue
	return c.StartDeferredTimeout(msg, timeout)
}
1211218
func NewMockBackoffFrom(i backoff.Backoff) *MockBackoff {
	return &MockBackoff{
		CloneFunc: &BackoffCloneFunc{
			defaultHook: i.Clone,
		},
		NextIntervalFunc: &BackoffNextIntervalFunc{
			defaultHook: i.NextInterval,
		},
		ResetFunc: &BackoffResetFunc{
			defaultHook: i.Reset,
		},
	}
}
1144386
func BackOff(attempts uint) int {
	upper := int(math.Min(float64(retryBackOffCap), float64(retryBackOffBase)*math.Pow(2.0, float64(attempts))))
	sleep := time.Duration(rand.Intn(upper)) * time.Millisecond
	time.Sleep(sleep)
	return int(sleep)
}
127821
func Requeue(id MessageID, delay time.Duration) *Command {
	var params = [][]byte{id[:], []byte(strconv.Itoa(int(delay / time.Millisecond)))}
	return &Command{[]byte("REQ"), params, nil}
}
398455
func (e *ExponentialStrategy) Backoff() backoff.Backoff {
	return &exponentialBackoff{
		first: e.opts.first,
		max:   e.opts.max.Nanoseconds(),
		rand:  e.opts.newRand(),
	}
}
380386
func (chainAsyncResult *ChainAsyncResult) GetWithTimeout(timeoutDuration, sleepDuration time.Duration) ([]reflect.Value, error) {
	if chainAsyncResult.backend == nil {
		return nil, ErrBackendNotConfigured
	}

	var (
		results []reflect.Value
		err     error
	)

	timeout := time.NewTimer(timeoutDuration)
	ln := len(chainAsyncResult.asyncResults)
	lastResult := chainAsyncResult.asyncResults[ln-1]

	for {
		select {
		case <-timeout.C:
			return nil, ErrTimeoutReached
		default:

			for _, asyncResult := range chainAsyncResult.asyncResults {
				_, errcur := asyncResult.Touch()
				if errcur != nil {
					return nil, err
				}
			}

			results, err = lastResult.Touch()
			if err != nil {
				return nil, err
			}
			if results != nil {
				return results, err
			}
			time.Sleep(sleepDuration)
		}
	}
}
278678
func (s *RegionRequestSender) SendReq(bo *Backoffer, req *tikvrpc.Request, regionID RegionVerID, timeout time.Duration) (*tikvrpc.Response, error) {
	resp, _, err := s.SendReqCtx(bo, req, regionID, timeout)
	return resp, err
}
185740
func (p *PriorityQueue) isPodBackingOff(pod *v1.Pod) bool {
	boTime, exists := p.podBackoff.GetBackoffTime(nsNameForPod(pod))
	if !exists {
		return false
	}
	return boTime.After(p.clock.Now())
}
1366309
func WithBackoffConfig(b BackoffConfig) DialOption {
	return withBackoff(backoff.Exponential{
		MaxDelay: b.MaxDelay,
	})
}
284978
func (c *Client) SetBackoff(backoff retryablehttp.Backoff) {
	c.modifyLock.RLock()
	c.config.modifyLock.Lock()
	defer c.config.modifyLock.Unlock()
	c.modifyLock.RUnlock()

	c.config.Backoff = backoff
}
420348
func RetryBackoff(retry int, minBackoff, maxBackoff time.Duration) time.Duration {
	if retry < 0 {
		retry = 0
	}

	backoff := minBackoff << uint(retry)
	if backoff > maxBackoff || backoff < minBackoff {
		backoff = maxBackoff
	}

	if backoff == 0 {
		return 0
	}
	return time.Duration(rand.Int63n(int64(backoff)))
}
786193
func (rkq *RekeyQueueStandard) Enqueue(id tlf.ID) {
	rkq.mu.Lock()
	defer rkq.mu.Unlock()
	rkq.pendings[id] = true

	select {
	case rkq.queue <- id:
	default:
		// The queue is full; drop this one for now until the next
		// request to the server for more rekeys.
		rkq.log.Debug("Rekey queue is full; dropping %s", id)
	}
}
572658
func (q *Queue) Cancel() error {
	select {
	case <-q.cancelled:
		// already cancelled, no-op
		return nil
	default:
		// mark the queue as cancelled
		close(q.cancelled)
		// Close the Queue, that will wait for pending commands to drain
		// will unblock any callers waiting on q.Block
		return q.Close()
	}
}
315496
func backoffDuration(retryCount int) time.Duration {
	// Upper limit of delay at ~1 minute
	if retryCount > 7 {
		retryCount = 7
	}

	rand.Seed(time.Now().UnixNano())
	delay := (1 << uint(retryCount)) * (rand.Intn(150) + 500)
	return time.Duration(delay) * time.Millisecond
}