// Make space in the new buckets by expiring the really bad entries.
// If no bad entries are available we remove the oldest.
180220
func (a *addrBook) expireNew(bucketIdx int) {
	for addrStr, ka := range a.bucketsNew[bucketIdx] {
		// If an entry is bad, throw it away
		if ka.isBad() {
			a.Logger.Info(fmt.Sprintf("expiring bad address %v", addrStr))
			a.removeFromBucket(ka, bucketTypeNew, bucketIdx)
			return
		}
	}

	// If we haven't thrown out a bad entry, throw out the oldest entry
	oldest := a.pickOldest(bucketTypeNew, bucketIdx)
	a.removeFromBucket(oldest, bucketTypeNew, bucketIdx)
}
591271
func NewLimiterExpiringBuckets(max int64, ttl, bucketDefaultExpirationTTL, bucketExpireJobInterval time.Duration) *Limiter {
	limiter := NewLimiter(max, ttl).SetExpiring(bucketDefaultExpirationTTL, bucketExpireJobInterval)
	return limiter
}
612430
func (l *WAL) DeleteBucketRange(orgID, bucketID influxdb.ID, min, max int64) (int, error) {
	if !l.enabled {
		return -1, nil
	}

	entry := &DeleteBucketRangeWALEntry{
		OrgID:    orgID,
		BucketID: bucketID,
		Min:      min,
		Max:      max,
	}

	id, err := l.writeToLog(entry)
	if err != nil {
		return -1, err
	}
	return id, nil
}
1326864
func (t *BucketRateLimiter) Close() (err error) {
	t.cancel()
	return t.egCleaner.Wait()
}
333960
func NewExpiringBuckets(max int64, ttl, bucketDefaultExpirationTTL, bucketExpireJobInterval time.Duration) *config.Limiter {
	return config.NewLimiterExpiringBuckets(max, ttl, bucketDefaultExpirationTTL, bucketExpireJobInterval)
}
899628
func (st *Store) Cleanup() {
	st.e.Where("expires_unix < ?", util.TimeStampNow()).Delete(&xormSession{tableName: st.opts.TableName})
}
190762
func (a Aggregation) Buckets() []Bucket {
	result := []Bucket{}
	if buckets, ok := a["buckets"]; ok {
		for _, bucket := range buckets.([]interface{}) {
			result = append(result, bucket.(map[string]interface{}))
		}
	}

	return result
}
812567
func sortBuckets(buckets []float64) []float64 {
	for i, bucket := range buckets {
		if i < len(buckets)-1 && bucket >= buckets[i+1] {
			sort.Float64s(buckets)
			break
		}
	}

	return buckets
}
785708
func (s *ProxyServer) CleanupAndExit() error {
	encounteredError := userspace.CleanupLeftovers(s.IptInterface)
	encounteredError = iptables.CleanupLeftovers(s.IptInterface) || encounteredError
	encounteredError = ipvs.CleanupLeftovers(s.IpvsInterface, s.IptInterface, s.IpsetInterface, s.CleanupIPVS) || encounteredError
	if encounteredError {
		return errors.New("encountered an error while tearing down rules")
	}

	return nil
}
1171490
func New(
	ctx context.Context,
	limit Limit,
	burst int,
	options ...interface{},
) (limiter *BucketRateLimiter) {
	ctx, cancel := context.WithCancel(ctx)
	cleanTimeout := time.Minute
	tickerInterval := time.Minute

	for _, option := range options {
		switch opt := option.(type) {
		case OptionCleanTimeout:
			cleanTimeout = time.Duration(opt)
		case OptionCleanTicker:
			tickerInterval = time.Duration(opt)
		}
	}

	egCleaner, _ := errgroup.WithContext(ctx)
	limiter = &BucketRateLimiter{
		ctx:            ctx,
		cancel:         cancel,
		limit:          limit,
		burst:          burst,
		cleanTimeout:   cleanTimeout,
		tickerInterval: tickerInterval,

		buckets:   map[string]*limiterWrap{},
		egCleaner: egCleaner,
		mutex:     sync.Mutex{},
	}
	egCleaner.Go(limiter.cleanerLoop)
	return
}
701040
func (b *bucket) Close() error {
	var err error

	// If the items were updated, write them out to disk.
	if b.itemsUpdated {
		err = writeBucket(b.file, b.items)
		if err != nil {
			b.backend.log.Warnf("Unable to update keys in %v: %v.", b.file.Name(), err)
		}
	}

	err = utils.FSUnlock(b.file)
	if err != nil {
		b.backend.log.Warnf("Unable to unlock file: %v.", err)
	}

	err = b.file.Close()
	if err != nil {
		b.backend.log.Warnf("Unable to close file: %v.", err)
	}

	return nil
}
1105168
func (f *Attempt_Data_Finished_) PurgeTimestamps() {
	if f.Finished.GetData() == nil {
		return
	}
	f.Finished.Data.Expiration = nil
}
74996
func (p *Pool) Cleanup() {
	p.Stop()
	p.lock.Lock()
	defer p.lock.Unlock()
	p.baseCancel()
}
1164498
func (agg *HealthD) purge() {
	var threshold = agg.intervalDuration * 5 // NOTE: this is arbitrary.
	for k, _ := range agg.hostAggregations {
		if time.Since(k.Time) > threshold {
			delete(agg.hostAggregations, k)
		}
	}

	n := len(agg.intervalAggregations)
	if n > agg.maxIntervals {
		agg.intervalAggregations = agg.intervalAggregations[(n - agg.maxIntervals):]
	}
}
54204
func (b *Bucket) Add(k string, exp int, v interface{}) (added bool, err error) {
	err = b.Write(k, 0, exp, v, AddOnly)
	if err == ErrKeyExists {
		return false, nil
	}
	return (err == nil), err
}
1152396
func (s *KVStore) Buckets(ctx context.Context) [][]byte {
	s.mu.RLock()
	defer s.mu.RUnlock()

	buckets := make([][]byte, 0, len(s.buckets))
	for b := range s.buckets {
		buckets = append(buckets, []byte(b))
	}
	return buckets
}
766545
func newBucket(cap int, period time.Duration) (b *Bucket) {
	b = &Bucket{
		cap:         cap,
		channelQuit: make(chan bool),
		count:       0,
		period:      period,
	}
	go b.tick()
	return
}