// 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
763733
func (c *ExpireCache) 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 {
			c.mutex.Lock()
			record, ok := c.cache[key]
			c.mutex.Unlock()
			if !ok {
				return errors.Errorf("Each() - key '%+v' disapeared "+
					"from cache during iteration", key)
			}

			err := callBack(key, record.Value)
			if err != nil {
				return err
			}

			c.mutex.Lock()
			if record.ExpireAt.Before(time.Now().UTC()) {
				delete(c.cache, key)
			}
			c.mutex.Unlock()
			return nil
		}, key)
	}

	// Wait for all the routines to complete
	errs := fanOut.Wait()
	if errs != nil {
		return errs
	}

	return nil
}
732842
func (prc *CorrectionProcessor) Process(nim image.Image, params imageserver.Params) (image.Image, error) {
	enabled, err := prc.isEnabled(params)
	if err != nil {
		return nil, err
	}
	if enabled {
		return prc.process(nim, params)
	}
	return prc.Processor.Process(nim, params)
}
165088
func (bg *Reader) cachePut(b Block) (evicted Block, retained bool) {
	if b == nil || !b.hasData() {
		return b, false
	}
	return bg.cache.Put(b)
}
660993
func (cache *Cache) Clean() {
	now := time.Now().UnixNano()

	cache.mutex.Lock()
	for key, counter := range cache.counters {
		if now > counter.Expiration {
			delete(cache.counters, key)
		}
	}
	cache.mutex.Unlock()
}
717099
func (c *Cache) pruneSizeLocked() {
	// There is no size constraint.
	if c.size <= 0 {
		return
	}

	// Prune the oldest entries until we're within our size constraint.
	for e := c.lru.Back(); e != nil; e = c.lru.Back() {
		if len(c.cache) <= c.size {
			return
		}
		c.evictEntryLocked(e)
	}
}
896625
func (c *Cache) Items() []Item {
	c.Lock()
	defer c.Unlock()
	items := make([]Item, 0, c.l.Len())
	for elem := c.l.Front(); elem != nil; elem = elem.Next() {
		items = append(items, *elem.Value.(*Item))
	}
	return items
}
314444
func (dbp *Process) ThreadList() []proc.Thread {
	r := make([]proc.Thread, 0, len(dbp.threads))
	for _, v := range dbp.threads {
		r = append(r, v)
	}
	return r
}
245844
func (c *InMemoryCache) Clear() {
	c.items.Range(func(key, _ interface{}) bool {
		c.items.Delete(key)
		return true
	})
}
715395
func RowProcessing(rows *core.Rows, fields []string, fn func(data *reflect.Value, index int, fieldName string) error) (err error) {
	length := len(fields)
	scanResultContainers := make([]interface{}, length)
	for i := 0; i < length; i++ {
		var resultContainer interface{}
		scanResultContainers[i] = &resultContainer
	}
	if err := rows.Scan(scanResultContainers...); err != nil {
		return err
	}
	for ii, key := range fields {
		rawValue := reflect.Indirect(reflect.ValueOf(scanResultContainers[ii]))
		fn(&rawValue, ii, key)
	}
	return nil
}
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
}
472066
func process(workload Workload, result *Result) (err error) {

	c := make(chan int, len(workload.Requests))
	result.Responses = make([]Response, len(workload.Requests))

	for index, _ := range workload.Requests {

		if workload.UseHeaders {
			replaceHeaderValues(&workload.Requests[index].Header, &workload.header)
		}

		if workload.StrictOrder {
			syncRequest(&workload.Requests[index], &result.Responses[index])
		} else {
			go asyncRequest(&workload.Requests[index], &result.Responses[index], c)
		}
	}

	// if async, wait for our responses or timeout
	if !workload.StrictOrder {

		// create a timeout so we don't wait forever
		var timeout <-chan time.Time

		if workload.Timeout > 0 {
			timeout = time.After(time.Duration(workload.Timeout))
		} else {
			timeout = time.After(DefaultTimeout)
		}

		for index := 0; index < len(workload.Requests); index++ {
			select {
			case <-c:
			case <-timeout:
				log.Warn("[process] Timed out waiting for all go routines to complete")
				break
			}
		}
	}
	return
}
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
}
1217518
func (f ProcessorFunc) Process(nim image.Image, params imageserver.Params) (image.Image, error) {
	return f(nim, params)
}
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)
		}
	}
}
40506
func ProcessEvents(c cache.ThreadSafeStore, indexer trigger.Indexer, queue workqueue.RateLimitingInterface, tags trigger.TagRetriever) cache.ResourceEventHandler {
	return cache.ResourceEventHandlerFuncs{
		AddFunc: func(obj interface{}) {
			key, entry, _, err := indexer.Index(obj, nil)
			if err != nil {
				utilruntime.HandleError(fmt.Errorf("unable to extract cache data from %T: %v", obj, err))
				return
			}
			if entry != nil {
				c.Add(key, entry)
				queue.Add(key)
			}
		},
		UpdateFunc: func(oldObj, newObj interface{}) {
			key, entry, change, err := indexer.Index(newObj, oldObj)
			if err != nil {
				utilruntime.HandleError(fmt.Errorf("unable to extract cache data from %T: %v", newObj, err))
				return
			}
			switch {
			case entry == nil:
				c.Delete(key)
			case change == cache.Added:
				c.Add(key, entry)
				queue.Add(key)
			case change == cache.Updated:
				c.Update(key, entry)
				queue.Add(key)
			}
		},
		DeleteFunc: func(obj interface{}) {
			key, entry, _, err := indexer.Index(nil, obj)
			if err != nil {
				utilruntime.HandleError(fmt.Errorf("unable to extract cache data from %T: %v", obj, err))
				return
			}
			if entry != nil {
				c.Delete(key)
			}
		},
	}
}
1340479
func (j *jws) cache() (err error) {
	if !j.clean {
		j.plcache, err = j.payload.Base64()
		j.clean = err == nil
	}
	return err
}
398351
func (o *cacheOnce) cache(f func() error) {
	if atomic.LoadUint32(&o.done) == 1 {
		return
	}
	// Slow-path.
	o.m.Lock()
	defer o.m.Unlock()
	if o.done == 0 {
		err := f()
		if err == nil {
			atomic.StoreUint32(&o.done, 1)
		}
	}
}