// A helper used to specify a CasMulti mutation operation on a shard client.
347200
func casMultiMutator(shardClient Client, mapping *ShardMapping) []MutateResponse {
	return shardClient.CasMulti(mapping.Items)
}
1193886
func (i *Int64) CAS(old, new int64) bool {
	return atomic.CompareAndSwapInt64(&i.v, old, new)
}
1216634
func (e *executor) mapperLocal(ctx context.Context, shards []uint64, mapFn mapFunc, reduceFn reduceFunc) (interface{}, error) {
	span, ctx := tracing.StartSpanFromContext(ctx, "Executor.mapperLocal")
	defer span.Finish()

	ch := make(chan mapResponse, len(shards))

	for _, shard := range shards {
		go func(shard uint64) {
			result, err := mapFn(shard)

			// Return response to the channel.
			select {
			case <-ctx.Done():
			case ch <- mapResponse{result: result, err: err}:
			}
		}(shard)
	}

	// Reduce results
	var maxShard int
	var result interface{}
	for {
		select {
		case <-ctx.Done():
			return nil, ctx.Err()
		case resp := <-ch:
			if resp.err != nil {
				return nil, resp.err
			}
			result = reduceFn(result, resp.result)
			maxShard++
		}

		// Exit once all shards are processed.
		if maxShard == len(shards) {
			return result, nil
		}
	}
}
960989
func (api *API) ShardNodes(ctx context.Context, indexName string, shard uint64) ([]*Node, error) {
	span, _ := tracing.StartSpanFromContext(ctx, "API.ShardNodes")
	defer span.Finish()

	if err := api.validate(apiShardNodes); err != nil {
		return nil, errors.Wrap(err, "validating api method")
	}

	return api.cluster.shardNodes(indexName, shard), nil
}
1138541
func (c *RawBinaryClient) mutate(code opCode, item *Item) MutateResponse {
	if item == nil {
		return NewMutateErrorResponse("", errors.New("item is nil"))
	}

	c.mutex.Lock()
	defer c.mutex.Unlock()

	if resp := c.sendMutateRequest(code, item, true); resp != nil {
		return resp
	}

	return c.receiveMutateResponse(code, item.Key)
}
1093924
func (ts *Server) DeleteShardReplication(ctx context.Context, cell, keyspace, shard string) error {
	conn, err := ts.ConnForCell(ctx, cell)
	if err != nil {
		return err
	}

	nodePath := path.Join(KeyspacesPath, keyspace, ShardsPath, shard, ShardReplicationFile)
	return conn.Delete(ctx, nodePath, nil)
}
1320936
func (e *executor) executeNotShard(ctx context.Context, index string, c *pql.Call, shard uint64) (*Row, error) {
	span, ctx := tracing.StartSpanFromContext(ctx, "Executor.executeNotShard")
	defer span.Finish()

	if len(c.Children) == 0 {
		return nil, errors.New("Not() requires an input row")
	} else if len(c.Children) > 1 {
		return nil, errors.New("Not() only accepts a single row input")
	}

	// Make sure the index supports existence tracking.
	idx := e.Holder.Index(index)
	if idx == nil {
		return nil, ErrIndexNotFound
	} else if idx.existenceField() == nil {
		return nil, errors.Errorf("index does not support existence tracking: %s", index)
	}

	var existenceRow *Row
	existenceFrag := e.Holder.fragment(index, existenceFieldName, viewStandard, shard)
	if existenceFrag == nil {
		existenceRow = NewRow()
	} else {
		existenceRow = existenceFrag.row(0)
	}

	row, err := e.executeBitmapCallShard(ctx, index, c.Children[0], shard)
	if err != nil {
		return nil, err
	}

	return existenceRow.Difference(row), nil
}
1349649
func (vc *vcursorImpl) StreamExecuteMulti(query string, rss []*srvtopo.ResolvedShard, bindVars []map[string]*querypb.BindVariable, callback func(reply *sqltypes.Result) error) error {
	atomic.AddUint32(&vc.logStats.ShardQueries, uint32(len(rss)))
	return vc.executor.scatterConn.StreamExecuteMulti(vc.ctx, vc.marginComments.Leading+query+vc.marginComments.Trailing, rss, bindVars, vc.tabletType, vc.safeSession.Options, callback)
}
972551
func (c *MockClient) DeleteMulti(keys []string) []MutateResponse {
	res := make([]MutateResponse, len(keys))
	for i, key := range keys {
		res[i] = c.Delete(key)
	}
	return res
}
322347
func (c *RedisClient) AddMulti(rs []AddRequest) ([]string, error) {
	for _, r := range rs {
		if err := c.conn.Send("ADDJOB", addArgs(r)...); err != nil {
			return nil, err
		}
	}
	// flush the output buffer and receive pending replies
	replies, err := redis.Values(c.conn.Do(""))
	ids := make([]string, len(replies))
	for i, val := range replies {
		if id, ok := val.(string); ok {
			ids[i] = id
		}
	}
	return ids, err
}
1195610
func (m ConcurrentMap) IterCb(fn IterCb) {
	for idx := range m {
		shard := (m)[idx]
		shard.RLock()
		for key, value := range shard.items {
			fn(key, value)
		}
		shard.RUnlock()
	}
}
497551
func (m *BaseShardManager) Init(
	shardFunc func(key string, numShard int) (shard int),
	logError func(err error),
	logInfo func(v ...interface{}),
	options net2.ConnectionOptions) {

	m.InitWithPool(
		shardFunc,
		logError,
		logInfo,
		net2.NewMultiConnectionPool(options))
}
134177
func (e *executor) executeBitmapCallShard(ctx context.Context, index string, c *pql.Call, shard uint64) (*Row, error) {
	if err := validateQueryContext(ctx); err != nil {
		return nil, err
	}

	span, ctx := tracing.StartSpanFromContext(ctx, "Executor.executeBitmapCallShard")
	defer span.Finish()

	switch c.Name {
	case "Row", "Range":
		return e.executeRowShard(ctx, index, c, shard)
	case "Difference":
		return e.executeDifferenceShard(ctx, index, c, shard)
	case "Intersect":
		return e.executeIntersectShard(ctx, index, c, shard)
	case "Union":
		return e.executeUnionShard(ctx, index, c, shard)
	case "Xor":
		return e.executeXorShard(ctx, index, c, shard)
	case "Not":
		return e.executeNotShard(ctx, index, c, shard)
	case "Shift":
		return e.executeShiftShard(ctx, index, c, shard)
	default:
		return nil, fmt.Errorf("unknown call: %s", c.Name)
	}
}
926180
func (e *executor) executeShiftShard(ctx context.Context, index string, c *pql.Call, shard uint64) (*Row, error) {
	n, _, err := c.IntArg("n")
	if err != nil {
		return nil, fmt.Errorf("executeShiftShard: %v", err)
	}

	if len(c.Children) == 0 {
		return nil, errors.New("Shift() requires an input row")
	} else if len(c.Children) > 1 {
		return nil, errors.New("Shift() only accepts a single row input")
	}

	row, err := e.executeBitmapCallShard(ctx, index, c.Children[0], shard)
	if err != nil {
		return nil, err
	}

	return row.Shift(n)
}
980691
func (c *MockClient) SetMulti(items []*Item) []MutateResponse {
	c.mutex.Lock()
	defer c.mutex.Unlock()

	res := make([]MutateResponse, len(items))
	for i, item := range items {
		res[i] = c.setHelper(item)
	}
	return res
}
836774
func (r *RedisDriver) GetMulti(cveIDs []string) (map[string]models.CveDetail, error) {
	cveDetails := map[string]models.CveDetail{}
	pipe := r.conn.Pipeline()
	cveRs, cpeRs := map[string]*redis.StringStringMapCmd{}, map[string]*redis.StringStringMapCmd{}
	for _, cveID := range cveIDs {
		cveRs[cveID] = pipe.HGetAll(hashKeyPrefix + cveID)
		cpeRs[cveID] = pipe.HGetAll(cpeHashKeyPrefix + cveID)
	}
	if _, err := pipe.Exec(); err != nil {
		if err != redis.Nil {
			return nil, fmt.Errorf("Failed to get multi cve json. err : %s", err)
		}
	}

	for cveID, cveResult := range cveRs {
		cpeResult := cpeRs[cveID]
		cveDetail, err := r.unmarshal(cveID, cveResult, cpeResult)
		if err != nil {
			return nil, err
		}
		cveDetails[cveID] = *cveDetail
	}
	return cveDetails, nil
}
501444
func (s *rowSegment) Union(other *rowSegment) *rowSegment {
	data := s.data.Union(&other.data)

	return &rowSegment{
		data:  *data,
		shard: s.shard,
		n:     data.Count(),
	}
}