// Iterates over all cached storage items first in cache and then in backend until consumer returns true for 'stop'
343573
func (cache *Cache) IterateCachedStorage(address crypto.Address,
	consumer func(key, value binary.Word256) error) error {
	accInfo, err := cache.get(address)
	if err != nil {
		return err
	}
	accInfo.RLock()
	// Try cache first for early exit
	for key, value := range accInfo.storage {
		if err := consumer(key, value); err != nil {
			accInfo.RUnlock()
			return err
		}
	}
	accInfo.RUnlock()
	return err
}
1217234
func (_m *LiveStateCache) IterateHierarchy(server string, key kube.ResourceKey, action func(v1alpha1.ResourceNode)) error {
	ret := _m.Called(server, key, action)

	var r0 error
	if rf, ok := ret.Get(0).(func(string, kube.ResourceKey, func(v1alpha1.ResourceNode)) error); ok {
		r0 = rf(server, key, action)
	} else {
		r0 = ret.Error(0)
	}

	return r0
}
95548
func (s *CacheOnStorage) Flush() error {
	if err := s.storage.Flush(); err != nil {
		return err
	}

	return s.cache.Flush()
}
1114710
func (store *KapacitorStore) All(ctx context.Context) ([]chronograf.Server, error) {
	if store.Kapacitor != nil {
		return []chronograf.Server{*store.Kapacitor}, nil
	}
	return nil, nil
}
1254237
func (e *Eventer) iterateConsumers(event *drivers.TaskEvent) {
	e.consumersLock.Lock()
	filtered := e.consumers[:0]
	for _, consumer := range e.consumers {

		// prioritize checking if context is cancelled prior
		// to attempting to forwarding events
		// golang select evaluations aren't predictable
		if consumer.ctx.Err() != nil {
			close(consumer.ch)
			continue
		}

		select {
		case <-time.After(consumer.timeout):
			filtered = append(filtered, consumer)
			e.logger.Warn("timeout sending event", "task_id", event.TaskID, "message", event.Message)
		case <-consumer.ctx.Done():
			// consumer context finished, filtering it out of loop
			close(consumer.ch)
		case consumer.ch <- event:
			filtered = append(filtered, consumer)
		}
	}
	e.consumers = filtered
	e.consumersLock.Unlock()
}
717905
func (h *hawkularSink) cache(md *metrics.MetricDefinition) {
	h.pushToCache(md.ID, hashDefinition(md))
}
555611
func (c *ServerCommand) storageMigrationActive(backend physical.Backend) bool {
	first := true

	for {
		migrationStatus, err := CheckStorageMigration(backend)
		if err == nil {
			if migrationStatus != nil {
				startTime := migrationStatus.Start.Format(time.RFC3339)
				c.UI.Error(wrapAtLength(fmt.Sprintf("ERROR! Storage migration in progress (started: %s). "+
					"Server startup is prevented until the migration completes. Use 'vault operator migrate -reset' "+
					"to force clear the migration lock.", startTime)))
				return true
			}
			return false
		}
		if first {
			first = false
			c.UI.Warn("\nWARNING! Unable to read storage migration status.")

			// unexpected state, so stop buffering log messages
			c.logGate.Flush()
		}
		c.logger.Warn("storage migration check error", "error", err.Error())

		select {
		case <-time.After(2 * time.Second):
		case <-c.ShutdownCh:
			return true
		}
	}
}
1191405
func (sp *StoragePlugin) Stop(ctx context.Context) {
	sp.stopOnce.Do(func() {
		sp.server.Stop()
		log.Info("stopped")
	})
}
354370
func (store *SourcesStore) All(ctx context.Context) ([]chronograf.Source, error) {
	if store.Source != nil {
		return []chronograf.Source{*store.Source}, nil
	}
	return nil, nil
}
444935
func (c *Cluster) StopStore(storeID uint64) {
	c.Lock()
	defer c.Unlock()

	if store := c.stores[storeID]; store != nil {
		store.meta.State = metapb.StoreState_Offline
	}
}
1226218
func (ListServiceOfferings) Each(resp interface{}, callback IterateItemFunc) {
	items, ok := resp.(*ListServiceOfferingsResponse)
	if !ok {
		callback(nil, fmt.Errorf("wrong type, ListServiceOfferingsResponse was expected, got %T", resp))
		return
	}

	for i := range items.ServiceOffering {
		if !callback(&items.ServiceOffering[i], nil) {
			break
		}
	}
}
550955
func (c LRUCache) Each(concurrent int, callBack func(key interface{}, value interface{}) error) []error {
	fanOut := NewFanOut(concurrent)
	keys := c.Keys()

	for _, key := range keys {
		fanOut.Run(func(key interface{}) error {
			value, ok := c.Peek(key)
			if !ok {
				// Key disappeared during cache iteration, This can occur as
				// expiration and removal can happen during iteration
				return nil
			}

			err := callBack(key, value)
			if err != nil {
				return err
			}
			return nil
		}, key)
	}

	// Wait for all the routines to complete
	errs := fanOut.Wait()
	if errs != nil {
		return errs
	}
	return nil
}
217116
func (sb *storageBackend) AllStorageInstances() ([]StorageInstance, error) {
	storageInstances, err := sb.storageInstances(nil)
	if err != nil {
		return nil, errors.Trace(err)
	}
	out := make([]StorageInstance, len(storageInstances))
	for i, s := range storageInstances {
		out[i] = s
	}
	return out, nil
}
556222
func GetStorageBackend(ctx context.Context) StorageBackend {
	be, ok := ctx.Value(ctxKeyStorageBackend).(StorageBackend)
	if !ok {
		return FS
	}
	return be
}
292291
func (c *Cache) DeleteAll() {
	var keys []interface{}
	c.m.Range(func(key interface{}, value interface{}) bool {
		c.close(key.(string), value.(future))
		keys = append(keys, key)
		return true
	})

	for _, key := range keys {
		c.m.Delete(key)
	}
}
650435
func (p *BlockPrefetchClient) Stop() {
	p.Lock()
	close(p.retargetChan)
	p.Unlock()
}
1339813
func (q *queue) countProcessableItems() int {
	for i, result := range q.resultCache {
		if result == nil || result.Pending > 0 {
			return i
		}
	}
	return len(q.resultCache)
}