// Processes each item in the cache in a thread safe way, such that the cache can be in use
// while processing items in the cache. Processing the cache with `Each()` does not update
// the expiration or last used.
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
}
917654
func (prc ListProcessor) Process(nim image.Image, params imageserver.Params) (image.Image, error) {
	for _, p := range prc {
		var err error
		nim, err = p.Process(nim, params)
		if err != nil {
			return nil, err
		}
	}
	return nim, nil
}
203077
func (ListTemplates) Each(resp interface{}, callback IterateItemFunc) {
	items, ok := resp.(*ListTemplatesResponse)
	if !ok {
		callback(nil, fmt.Errorf("wrong type, ListTemplatesResponse was expected, got %T", resp))
		return
	}

	for i := range items.Template {
		if !callback(&items.Template[i], nil) {
			break
		}
	}
}
445012
func (m *MemoryStorer) Clean() {
	t := time.Now().UTC()
	m.mut.Lock()
	for id, session := range m.sessions {
		if t.After(session.expires) {
			delete(m.sessions, id)
		}
	}
	m.mut.Unlock()
}
897020
func (c *Cache) Prune(ctx context.Context) {
	now := clock.Now(ctx)

	c.lock.Lock()
	defer c.lock.Unlock()

	c.pruneExpiryLocked(now)
}
880990
func (ListTags) Each(resp interface{}, callback IterateItemFunc) {
	items, ok := resp.(*ListTagsResponse)
	if !ok {
		callback(nil, fmt.Errorf("wrong type, ListTagsResponse was expected, got %T", resp))
		return
	}

	for i := range items.Tag {
		if !callback(&items.Tag[i], nil) {
			break
		}
	}
}
1182624
func (c *cache) cleanup() error {
	unused := []string{}
	now := time.Now()
	// First find old ones.
	for id, bucket := range c.buckets {
		if bucket.status == statusLoading {
			continue
		}
		if bucket.lastUsed.Add(c.ttl).Before(now) {
			unused = append(unused, id)
		}
	}
	// Now delete found ones.
	var errs []error
	for _, id := range unused {
		cacheable := c.buckets[id].cacheable
		delete(c.buckets, id)
		if err := cacheable.Discard(); err != nil {
			errs = append(errs, err)
		}
	}
	if len(errs) > 0 {
		return errors.Collect(errs...)
	}
	return nil
}
104468
func (ListUsers) Each(resp interface{}, callback IterateItemFunc) {
	items, ok := resp.(*ListUsersResponse)
	if !ok {
		callback(nil, fmt.Errorf("wrong type, ListUsersResponse was expected, got %T", resp))
		return
	}

	for i := range items.User {
		if !callback(&items.User[i], nil) {
			break
		}
	}
}
219947
func (ListInstanceGroups) Each(resp interface{}, callback IterateItemFunc) {
	items, ok := resp.(*ListInstanceGroupsResponse)
	if !ok {
		callback(nil, fmt.Errorf("wrong type, ListInstanceGroupsResponse was expected, got %T", resp))
		return
	}

	for i := range items.InstanceGroup {
		if !callback(&items.InstanceGroup[i], nil) {
			break
		}
	}
}
518082
func (ListNics) Each(resp interface{}, callback IterateItemFunc) {
	items, ok := resp.(*ListNicsResponse)
	if !ok {
		callback(nil, fmt.Errorf("wrong type, ListNicsResponse was expected, got %T", resp))
		return
	}

	for i := range items.Nic {
		if !callback(&items.Nic[i], nil) {
			break
		}
	}
}
832903
func (l *Synchronizer) PostProcess(processing map[string]struct{}) {
	// Get all locally stored data
	nodes, err := l.GetAll()
	if err != nil {
		glog.Warningf("Could not access locally stored data: %s", err)
		return
	}

	// Delete any local nodes that do not exist remotely and are not in process
	for _, node := range nodes {
		if _, ok := processing[node.GetID()]; ok {
			// pass
		} else if err := l.Delete(node.GetID()); err != nil {
			glog.Warningf("Could not delete %s from locally stored data: %s", node.GetID(), err)
		}
	}
}
1367630
func (c *InMemoryCache) Clear() {
	c.mu.Lock()
	c.items = map[string]Thunk{}
	c.mu.Unlock()
}
713030
func (cc *ConsolidatorCache) Items() []ConsolidatorCacheItem {
	items := cc.LRUCache.Items()
	ret := make([]ConsolidatorCacheItem, len(items))
	for i, v := range items {
		ret[i] = ConsolidatorCacheItem{Query: v.Key, Count: v.Value.(*ccount).get()}
	}
	return ret
}
35167
func Each(array []interface{}, iterator Iterator) {
	for index, data := range array {
		iterator(data, index)
	}
}
137417
func (ListResourceDetails) Each(resp interface{}, callback IterateItemFunc) {
	items, ok := resp.(*ListResourceDetailsResponse)
	if !ok {
		callback(nil, fmt.Errorf("wrong type, ListResourceDetailsResponse was expected, got %T", resp))
		return
	}

	for i := range items.ResourceDetail {
		if !callback(&items.ResourceDetail[i], nil) {
			break
		}
	}
}
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)
	}
}
1275235
func (ListISOs) Each(resp interface{}, callback IterateItemFunc) {
	items, ok := resp.(*ListISOsResponse)
	if !ok {
		callback(nil, fmt.Errorf("wrong type, ListISOsResponse was expected, got %T", resp))
		return
	}

	for i := range items.ISO {
		if !callback(&items.ISO[i], nil) {
			break
		}
	}
}