// AddShard add a list of shards to the cluster.
348370
func (c *Cluster) AddShard(addresses ...string) {
	if !c.sentinel {
		for _, address := range addresses {
			c.addresses = append(c.addresses, &addressWithPassword{address, ""})
		}
	}
}
808604
func (m ConcurrentMap) Upsert(key string, value interface{}, cb UpsertCb) (res interface{}) {
	shard := m.GetShard(key)
	shard.Lock()
	v, ok := shard.items[key]
	res = cb(ok, v, value)
	shard.items[key] = res
	shard.Unlock()
	return res
}
376677
func (m *Map) Add(keys ...string) {
	for _, key := range keys {
		for i := 0; i < m.replicas; i++ {
			hash := int(m.hash([]byte(strconv.Itoa(i) + key)))
			m.keys = append(m.keys, hash)
			m.hashMap[hash] = key
		}
	}
	sort.Ints(m.keys)
}
551774
func (i *Index) Add(r Record, c bgzf.Chunk, placed, mapped bool) error {
	refName := r.RefName()
	rid, ok := i.nameMap[refName]
	if !ok {
		rid = len(i.refNames)
		i.refNames = append(i.refNames, refName)
	}
	shim := tabixShim{id: rid, start: r.Start(), end: r.End()}
	return i.idx.Add(shim, internal.BinFor(r.Start(), r.End()), c, placed, mapped)
}
1016912
func (ch *CRLChecker) shard(c context.Context, idx int) (shards.Shard, error) {
	val, err := ch.shards[idx].Get(c, func(prev interface{}) (interface{}, time.Duration, error) {
		prevState, _ := prev.(shardCache)
		newState, err := ch.refetchShard(c, idx, prevState)
		return newState, ch.cacheDuration, err
	})
	if err != nil {
		return nil, err
	}
	// lazyslot.Get always returns non-nil val on success. It is safe to cast it
	// to whatever we returned in the callback (which is always shardCache, see
	// refetchShard).
	return val.(shardCache).shard, nil
}
791000
func (s *DisableEnhancedMonitoringInput) SetShardLevelMetrics(v []*string) *DisableEnhancedMonitoringInput {
	s.ShardLevelMetrics = v
	return s
}
743710
func (m *BaseShardManager) GetShardsForKeys(
	keys []string) map[int]*ShardMapping {

	m.rwMutex.RLock()
	defer m.rwMutex.RUnlock()

	numShards := len(m.shardStates)
	results := make(map[int]*ShardMapping)

	for _, key := range keys {
		shardId := m.getShardId(key, numShards)

		entry, inMap := results[shardId]
		if !inMap {
			entry = &ShardMapping{}
			if shardId != -1 {
				state := m.shardStates[shardId]
				if state.State == ActiveServer {
					m.fillEntryWithConnection(state.Address, entry)
				} else {
					connSkippedByAddr.Add(state.Address, 1)
				}
			}
			entry.Keys = make([]string, 0, 1)
			results[shardId] = entry
		}
		entry.Keys = append(entry.Keys, key)
	}

	return results
}
487740
func (wr *Wrangler) SetSourceShards(ctx context.Context, keyspace, shard string, sources []*topodatapb.TabletAlias, tables []string) error {
	// Read the source tablets.
	sourceTablets, err := wr.ts.GetTabletMap(ctx, sources)
	if err != nil {
		return err
	}

	// Insert their KeyRange in the SourceShards array.
	// We use a linear 0-based id, that matches what worker/split_clone.go
	// inserts into _vt.vreplication.
	// We want to guarantee sourceShards[i] is using sources[i],
	// So iterating over the sourceTablets map would be a bad idea.
	sourceShards := make([]*topodatapb.Shard_SourceShard, len(sourceTablets))
	for i, alias := range sources {
		ti := sourceTablets[topoproto.TabletAliasString(alias)]
		sourceShards[i] = &topodatapb.Shard_SourceShard{
			Uid:      uint32(i),
			Keyspace: ti.Keyspace,
			Shard:    ti.Shard,
			KeyRange: ti.KeyRange,
			Tables:   tables,
		}
	}

	// Update the shard with the new source shards.
	_, err = wr.ts.UpdateShardFields(ctx, keyspace, shard, func(si *topo.ShardInfo) error {
		// If the shard already has sources, maybe it's already been restored,
		// so let's be safe and abort right here.
		if len(si.SourceShards) > 0 {
			return fmt.Errorf("shard %v/%v already has SourceShards, not overwriting them (full record: %v)", keyspace, shard, *si.Shard)
		}

		si.SourceShards = sourceShards
		return nil
	})
	return err
}
1169948
func (ch *CRLChecker) refetchShard(c context.Context, idx int, prevState shardCache) (newState shardCache, err error) {
	// Have something locally already? Quickly fetch CRLShardHeader to check
	// whether we need to pull a heavy CRLShardBody.
	hdr := CRLShardHeader{ID: shardEntityID(ch.cn, ch.shardCount, idx)}
	if prevState.sha1 != "" {
		switch err = ds.Get(c, &hdr); {
		case err == ds.ErrNoSuchEntity:
			err = fmt.Errorf("shard header %q is missing", hdr.ID)
			return
		case err != nil:
			err = transient.Tag.Apply(err)
			return
		}
		// The currently cached copy is still good enough?
		if hdr.SHA1 == prevState.sha1 {
			newState = prevState
			return
		}
	}

	// Nothing is cached, or the datastore copy is fresher than what we have in
	// the cache. Need to fetch a new copy, unzip and deserialize it. This entity
	// is prepared by updateCRLShard.
	body := CRLShardBody{Parent: ds.KeyForObj(c, &hdr)}
	switch err = ds.Get(c, &body); {
	case err == ds.ErrNoSuchEntity:
		err = fmt.Errorf("shard body %q is missing", hdr.ID)
		return
	case err != nil:
		err = transient.Tag.Apply(err)
		return
	}

	// Unzip and deserialize.
	blob, err := utils.ZlibDecompress(body.ZippedData)
	if err != nil {
		return
	}
	shard, err := shards.ParseShard(blob)
	if err != nil {
		return
	}

	newState = shardCache{shard: shard, sha1: body.SHA1}
	return
}
367142
func (b *broker) listShards() ([]*kinesis.Shard, error) {
	var ss []*kinesis.Shard
	var listShardsInput = &kinesis.ListShardsInput{
		StreamName: aws.String(b.streamName),
	}

	for {
		resp, err := b.client.ListShards(listShardsInput)
		if err != nil {
			return nil, fmt.Errorf("ListShards error: %v", err)
		}
		ss = append(ss, resp.Shards...)

		if resp.NextToken == nil {
			return ss, nil
		}

		listShardsInput = &kinesis.ListShardsInput{
			NextToken:  resp.NextToken,
			StreamName: aws.String(b.streamName),
		}
	}
}
1010346
func (p *Plan) AddStats(execCount uint64, execTime time.Duration, shardQueries, rows, errors uint64) {
	p.mu.Lock()
	p.ExecCount += execCount
	p.ExecTime += execTime
	p.ShardQueries += shardQueries
	p.Rows += rows
	p.Errors += errors
	p.mu.Unlock()
}
966181
func (d *DiskEditValue) AddSSHKeys(keyID string) {
	if d.SSHKeys == nil {
		d.SSHKeys = []*SSHKey{}
	}
	d.SSHKeys = append(d.SSHKeys, &SSHKey{Resource: NewResourceByStringID(keyID)})
}
1247156
func (s *StreamDescription) SetShards(v []*Shard) *StreamDescription {
	s.Shards = v
	return s
}
678846
func (s *GetShardIteratorInput) SetShardId(v string) *GetShardIteratorInput {
	s.ShardId = &v
	return s
}
1251282
func (c *Cache) Add(index uint64, obj interface{}) bool {
	return c.getShard(index).add(index, obj)
}
1128091
func (conn *FakeVTGateConn) AddShardQuery(
	sql string,
	bindVariables map[string]*querypb.BindVariable,
	keyspace string,
	shards []string,
	tabletType topodatapb.TabletType,
	session *vtgatepb.Session,
	notInTransaction bool,
	expectedResult *sqltypes.Result) {
	conn.execMap[getShardQueryKey(sql, shards)] = &queryResponse{
		shardQuery: &queryExecuteShards{
			SQL:              sql,
			BindVariables:    bindVariables,
			Keyspace:         keyspace,
			Shards:           shards,
			TabletType:       tabletType,
			Session:          session,
			NotInTransaction: notInTransaction,
		},
		reply: expectedResult,
	}
}
151395
func (h *HLL) AddWithHasher(value string, hasher func(string) uint64) {
	hash := hasher(value)
	h.AddHash(hash)
}