// getOrCreateFetcher attempts at retrieving an existing fetchers
// if none exists, creates one and saves it in the fetchers cache
// caller must hold the lock
963317
func (n *NetStore) getOrCreateFetcher(ctx context.Context, ref Address) *fetcher {
	if f := n.getFetcher(ref); f != nil {
		return f
	}

	// no fetcher for the given address, we have to create a new one
	key := hex.EncodeToString(ref)
	// create the context during which fetching is kept alive
	cctx, cancel := context.WithTimeout(ctx, fetcherTimeout)
	// destroy is called when all requests finish
	destroy := func() {
		// remove fetcher from fetchers
		n.fetchers.Remove(key)
		// stop fetcher by cancelling context called when
		// all requests cancelled/timedout or chunk is delivered
		cancel()
	}
	// peers always stores all the peers which have an active request for the chunk. It is shared
	// between fetcher and the NewFetchFunc function. It is needed by the NewFetchFunc because
	// the peers which requested the chunk should not be requested to deliver it.
	peers := &sync.Map{}

	cctx, sp := spancontext.StartSpan(
		cctx,
		"netstore.fetcher",
	)

	sp.LogFields(olog.String("ref", ref.String()))
	fetcher := newFetcher(sp, ref, n.NewNetFetcherFunc(cctx, ref, peers), destroy, peers, n.closeC)
	n.fetchers.Add(key, fetcher)

	return fetcher
}
1188905
func (cp *CachePool) GetReader(fileIndex int64) (io.Reader, error) {
	// this will block until the channel is closed,
	// or immediately succeed if it's already closed
	<-cp.fileChans[fileIndex]

	return cp.cache.GetReader(fileIndex)
}
467531
func (q *Query) Prefetch(p float64) *Query {
	q.prefetch = p
	return q
}
1156525
func (c *LayeredCache) GetOrCreateSecondaryCache(primary string) *SecondaryCache {
	primaryBkt := c.bucket(primary)
	bkt := primaryBkt.getSecondaryBucket(primary)
	primaryBkt.Lock()
	if bkt == nil {
		bkt = &bucket{lookup: make(map[string]*Item)}
		primaryBkt.buckets[primary] = bkt
	}
	primaryBkt.Unlock()
	return &SecondaryCache{
		bucket: bkt,
		pCache: c,
	}
}
447532
func (c *cache) fetchClusters(ctx context.Context) (map[string]cluster.Cluster, error) {
	signer := newJWTSASigner(ctx, c.config, c.options...)
	cln, err := signer.createSignedClient()
	if err != nil {
		return nil, err
	}

	res, err := cln.ShowAuthClientClusters(goasupport.ForwardContextRequestID(ctx), clusterclient.ShowAuthClientClustersPath())
	if err != nil {
		return nil, err
	}
	defer rest.CloseResponse(res)

	if res.StatusCode != http.StatusOK {
		bodyString := rest.ReadBody(res.Body)
		log.Error(ctx, map[string]interface{}{
			"response_status": res.Status,
			"response_body":   bodyString,
		}, "unable to get clusters from Cluster Management Service")
		return nil, errors.Errorf("unable to get clusters from Cluster Management Service. Response status: %s. Response body: %s", res.Status, bodyString)
	}

	clusters, err := cln.DecodeFullClusterList(res)
	if err != nil {
		return nil, err
	}

	clusterMap := map[string]cluster.Cluster{}
	if clusters.Data != nil {
		for _, d := range clusters.Data {
			cls := cluster.Cluster{
				Name:                   d.Name,
				APIURL:                 d.APIURL,
				AppDNS:                 d.AppDNS,
				ConsoleURL:             d.ConsoleURL,
				LoggingURL:             d.LoggingURL,
				MetricsURL:             d.MetricsURL,
				TokenProviderID:        d.TokenProviderID,
				AuthClientDefaultScope: d.AuthClientDefaultScope,
				AuthClientID:           d.AuthClientID,
				AuthClientSecret:       d.AuthClientSecret,
				ServiceAccountUsername: d.ServiceAccountUsername,
				ServiceAccountToken:    d.ServiceAccountToken,
				CapacityExhausted:      d.CapacityExhausted,
			}
			clusterMap[rest.AddTrailingSlashToURL(cls.APIURL)] = cls
		}
	}
	return clusterMap, nil
}
337215
func newCacher(persister Persister) *cacher {
	return &cacher{
		persister: persister,
		cache:     make(map[string]*futures.Future, 10),
	}
}
1095060
func GetCreator() Creator {
	gc, ok := creators[*implementation]
	if !ok {
		log.Exitf("No gateway registered as %s", *implementation)
	}
	return gc
}
1272047
func (cb *CacheBuilder) LoaderFunc(loaderFunc LoaderFunc) *CacheBuilder {
	cb.loaderExpireFunc = func(k interface{}) (interface{}, *time.Duration, error) {
		v, err := loaderFunc(k)
		return v, nil, err
	}
	return cb
}
10364
func (m *MockCryptoSetup) GetSealer() (protocol.EncryptionLevel, handshake.Sealer) {
	m.ctrl.T.Helper()
	ret := m.ctrl.Call(m, "GetSealer")
	ret0, _ := ret[0].(protocol.EncryptionLevel)
	ret1, _ := ret[1].(handshake.Sealer)
	return ret0, ret1
}
627861
func NewBlockPrefetchClient(f BlockFetcher) *BlockPrefetchClient {
	p := &BlockPrefetchClient{
		f:            f,
		retargetChan: make(chan *chainhash.Hash, 10),
		current:      new(blockData),
		next:         new(blockData),
	}
	// Start the block fetcher goroutine, waiting for a signal on retargetChan.
	go p.fetcher()
	return p
}
1124915
func (k cursorKind) fetchPrevCursor(c context.Context, pkg, cursor string) string {
	itm, err := memcache.GetKey(c, k.cursorKey(pkg, cursor))
	if err != nil {
		if err != memcache.ErrCacheMiss {
			logging.WithError(err).Errorf(c, "Failed to fetch prev cursor %q from memcache", k)
		}
		return ""
	}
	return string(itm.Value())
}
1243257
func (s *SearchSource) FetchSource(fetchSource bool) *SearchSource {
	if s.fetchSourceContext == nil {
		s.fetchSourceContext = NewFetchSourceContext(fetchSource)
	} else {
		s.fetchSourceContext.SetFetchSource(fetchSource)
	}
	return s
}
1176468
func (c *ConnectCALeaf) fetchStart(rootUpdateCh chan struct{}) {
	c.rootWatchMu.Lock()
	defer c.rootWatchMu.Unlock()
	// Lazy allocation
	if c.rootWatchSubscribers == nil {
		c.rootWatchSubscribers = make(map[chan struct{}]struct{})
	}
	// Make sure a root watcher is running. We don't only do this on first request
	// to be more tolerant of errors that could cause the root watcher to fail and
	// exit.
	if c.rootWatchCancel == nil {
		ctx, cancel := context.WithCancel(context.Background())
		c.rootWatchCancel = cancel
		go c.rootWatcher(ctx)
	}
	c.rootWatchSubscribers[rootUpdateCh] = struct{}{}
}
749682
func (m *manager) GetAuthenticator() (auth.Authenticator, error) {
	return auth.NewJwtAuth(&auth.JwtAuthConfig{
		SharedSecret: []byte(m.config.SharedSecret),
	})
}
1192996
func (c *Changelog) GetVersionOrCreate(versionNum string) *Version {
	version := c.GetVersion(versionNum)
	if version == nil {
		version = NewVersion(versionNum)
		c.Versions = append([]*Version{version}, c.Versions...)
	}
	return version
}
960893
func (o *httpOps) AscRemoteFetcher() *remoteAscFetcher {
	ensureLogger(o.Debug)
	f := func(u *url.URL, file *os.File) error {
		switch u.Scheme {
		case "http", "https":
		default:
			return fmt.Errorf("invalid signature location: expected %q scheme, got %q", "http(s)", u.Scheme)
		}
		session := o.getSession(u, file, "signature", "")
		dl := o.getDownloader(session)
		err := dl.Download(u, file)
		if err != nil {
			return err
		}
		if session.Cd.UseCached {
			return fmt.Errorf("unexpected cache reuse request for signature %q", u.String())
		}
		return nil
	}
	return &remoteAscFetcher{
		F: f,
		S: o.S,
	}
}
1090046
func (b *Buffer) getOrCreateBuffer(keyspace, shard string) *shardBuffer {
	key := topoproto.KeyspaceShardString(keyspace, shard)
	b.mu.RLock()
	sb, ok := b.buffers[key]
	stopped := b.stopped
	b.mu.RUnlock()

	if stopped {
		return nil
	}
	if ok {
		return sb
	}

	b.mu.Lock()
	defer b.mu.Unlock()
	// Look it up again because it could have been created in the meantime.
	sb, ok = b.buffers[key]
	if !ok {
		sb = newShardBuffer(b.mode(keyspace, shard), keyspace, shard, b.now, b.bufferSizeSema)
		b.buffers[key] = sb
	}
	return sb
}