// BatchMulti sends a batch of payments from the user to multiple recipients in
// a single multi-operation transaction.
435163
func BatchMulti(mctx libkb.MetaContext, walletState *WalletState, arg stellar1.BatchLocalArg) (res stellar1.BatchResultLocal, err error) {
	mctx = mctx.WithLogTag("BMULT=" + arg.BatchID)

	startTime := time.Now()
	res.StartTime = stellar1.ToTimeMs(startTime)
	defer func() {
		if res.EndTime == 0 {
			res.EndTime = stellar1.ToTimeMs(time.Now())
		}
	}()

	// look up sender account
	senderAccountID, senderSeed, err := LookupSenderSeed(mctx)
	if err != nil {
		return res, err
	}

	mctx.Debug("Batch sender account ID: %s", senderAccountID)
	mctx.Debug("Batch size: %d", len(arg.Payments))

	results := make([]stellar1.BatchPaymentResult, len(arg.Payments))
	var multiOps []multiOp
	for i, payment := range arg.Payments {
		results[i] = stellar1.BatchPaymentResult{
			Username: libkb.NewNormalizedUsername(payment.Recipient).String(),
		}
		recipient, err := LookupRecipient(mctx, stellarcommon.RecipientInput(payment.Recipient), false /* isCLI for identify purposes */)
		if err != nil {
			mctx.Debug("LookupRecipient error: %s", err)
			makeResultError(&results[i], err)
			continue
		}

		if recipient.AccountID == nil {
			mop, err := prepareRelayOp(mctx, payment, recipient)
			if err != nil {
				makeResultError(&results[i], err)
				continue
			}
			multiOps = append(multiOps, mop)
		} else {
			mop, err := prepareDirectOp(mctx, walletState, payment, recipient)
			if err != nil {
				makeResultError(&results[i], err)
				continue
			}
			multiOps = append(multiOps, mop)
		}
	}

	baseFee := walletState.BaseFee(mctx)
	sp, unlock := NewSeqnoProvider(mctx, walletState)
	defer unlock()
	tx := stellarnet.NewBaseTx(stellarnet.AddressStr(senderAccountID), sp, baseFee)
	var post stellar1.PaymentMultiPost

	// add all the prepared ops here
	for _, mop := range multiOps {
		if mop.CreateAccount {
			tx.AddCreateAccountOp(stellarnet.AddressStr(mop.Recipient), mop.Amount)
		} else {
			tx.AddPaymentOp(stellarnet.AddressStr(mop.Recipient), mop.Amount)
		}
		post.Operations = append(post.Operations, mop.Op)
	}

	// sign the tx
	sr, err := tx.Sign(senderSeed)
	if err != nil {
		return res, err
	}
	post.SignedTransaction = sr.Signed
	post.FromDeviceID = mctx.ActiveDevice().DeviceID()

	// submit it
	submitRes, err := walletState.SubmitMultiPayment(mctx.Ctx(), post)
	if err != nil {
		// make all the results have an error
		for _, r := range results {
			makeResultError(&r, err)
		}
	} else {
		// make all ther results have success
		now := stellar1.ToTimeMs(time.Now())
		for i := 0; i < len(results); i++ {
			results[i].TxID = submitRes.TxID
			results[i].Status = stellar1.PaymentStatus_COMPLETED
			results[i].EndTime = now
		}
	}

	res.Payments = results

	return res, nil
}
16378
func (c *Client) MultipleBatch(operations []BatchOperationIndexed, opts ...interface{}) (res MultipleBatchRes, err error) {
	body := map[string][]BatchOperationIndexed{"requests": operations}
	path := c.path("/indexes/*/batch")
	err = c.transport.Request(&res, http.MethodPost, path, body, call.Write, opts...)
	res.wait = c.waitTask
	return
}
385333
func (c *RedisClient) GetMulti(count int, timeout time.Duration, queues ...string) ([]Job, error) {

	if len(queues) == 0 {
		return nil, errors.New("disque: no queues specified")
	}
	if count < 0 {
		return nil, fmt.Errorf("disque: invalid count %d", count)
	}

	args := redis.Args{}
	if timeout > 0 {
		args = args.Add("TIMEOUT", int64(timeout/time.Millisecond))
	}
	if count > 0 {
		args = args.Add("COUNT", count)
	}
	args = args.Add("FROM")
	args = args.AddFlat(queues)

	vals, err := redis.Values(c.conn.Do("GETJOB", args...))

	if err != nil {
		return nil, fmt.Errorf("disque: could not get jobs: %s", err)
	}

	ret := make([]Job, 0, len(vals))

	for _, v := range vals {
		if arr, ok := v.([]interface{}); ok {

			ret = append(ret, Job{
				Queue: string(arr[0].([]byte)),
				id:    string(arr[1].([]byte)),
				Data:  arr[2].([]byte),
			})

		}

	}

	return ret, nil

}
1343873
func (st *TxnState) BatchGet(keys []kv.Key) (map[string][]byte, error) {
	bufferValues := make([][]byte, len(keys))
	shrinkKeys := make([]kv.Key, 0, len(keys))
	for i, key := range keys {
		val, err := st.buf.Get(key)
		if kv.IsErrNotFound(err) {
			shrinkKeys = append(shrinkKeys, key)
			continue
		}
		if err != nil {
			return nil, err
		}
		if len(val) != 0 {
			bufferValues[i] = val
		}
	}
	storageValues, err := st.Transaction.BatchGet(shrinkKeys)
	if err != nil {
		return nil, err
	}
	for i, key := range keys {
		if bufferValues[i] == nil {
			continue
		}
		storageValues[string(key)] = bufferValues[i]
	}
	return storageValues, nil
}
990835
func (b *batchProducer) sendBatch(batchSize int) int {
	if len(b.records) == 0 {
		return 0
	}

	// In the future, maybe this could be a RetryPolicy or something
	if b.consecutiveErrors == 1 {
		b.currentDelay = 50 * time.Millisecond
	} else if b.consecutiveErrors > 1 {
		b.currentDelay *= 2
	}

	if b.currentDelay > 0 {
		b.logger.Printf("Delaying the batch by %v because of %v consecutive errors", b.currentDelay, b.consecutiveErrors)
		time.Sleep(b.currentDelay)
	}

	records := b.takeRecordsFromBuffer(batchSize)
	res, err := b.client.PutRecords(b.recordsToArgs(records))

	if err != nil {
		b.consecutiveErrors++
		b.currentStat.KinesisErrorsSinceLastStat++
		b.logger.Printf("Error occurred when sending PutRecords request to Kinesis stream %v: %v", b.streamName, err)

		if b.consecutiveErrors >= 5 && b.isBufferFullOrNearlyFull() {
			// In order to prevent Add from hanging indefinitely, we start dropping records
			b.logger.Printf("DROPPING %v records because buffer is full or nearly full and there have been %v consecutive errors from Kinesis", len(records), b.consecutiveErrors)
		} else {
			b.logger.Printf("Returning %v records to buffer (%v consecutive errors)", len(records), b.consecutiveErrors)
			// returnRecordsToBuffer can block if the buffer (channel) if full so we’ll
			// call it in a goroutine. This might be problematic WRT ordering. TODO: revisit this.
			go b.returnRecordsToBuffer(records)
		}

		return 0
	}

	b.consecutiveErrors = 0
	b.currentDelay = 0
	succeeded := len(records) - res.FailedRecordCount

	b.currentStat.RecordsSentSuccessfullySinceLastStat += succeeded

	if res.FailedRecordCount == 0 {
		b.logger.Printf("PutRecords request succeeded: sent %v records to Kinesis stream %v", succeeded, b.streamName)
	} else {
		b.logger.Printf("Partial success when sending a PutRecords request to Kinesis stream %v: %v succeeded, %v failed. Re-enqueueing failed records.", b.streamName, succeeded, res.FailedRecordCount)
		// returnSomeFailedRecordsToBuffer can block if the buffer (channel) if full so we’ll
		// call it in a goroutine. This might be problematic WRT ordering. TODO: revisit this.
		go b.returnSomeFailedRecordsToBuffer(res, records)
	}

	return succeeded
}
1154588
func executeBatches(batches []batch) error {
	logger.Debugf("Executing batches = %s", batches)
	numBatches := len(batches)
	if numBatches == 0 {
		return nil
	}
	if numBatches == 1 {
		return batches[0].execute()
	}
	var batchWG sync.WaitGroup
	batchWG.Add(numBatches)
	errsChan := make(chan error, numBatches)
	defer close(errsChan)
	for _, b := range batches {
		go func(b batch) {
			defer batchWG.Done()
			if err := b.execute(); err != nil {
				errsChan <- err
				return
			}
		}(b)
	}
	batchWG.Wait()
	if len(errsChan) > 0 {
		return <-errsChan
	}
	return nil
}
104768
func (db *DB) WriteBatch(batch *leveldb.Batch) (err error) {
	err = db.ldb.Write(batch, nil)
	if err != nil {
		metrics.GetOrRegisterCounter("DB.writebatchFail", nil).Inc(1)
		return err
	}
	metrics.GetOrRegisterCounter("DB.writebatch", nil).Inc(1)
	return nil
}
1303504
func (c *ShardedClient) GetMulti(keys []string) map[string]GetResponse {
	return c.getMulti(c.manager.GetShardsForKeys(keys))
}
890289
func (c *ShardedClient) DeleteMulti(keys []string) []MutateResponse {
	return c.mutateMulti(c.manager.GetShardsForKeys(keys), deleteMultiMutator)
}
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
}
612302
func (b *Bucket) InitMulti(ctx context.Context, key string, contType string, perm ACL) (*Multi, error) {
	headers := map[string][]string{
		"Content-Type":   {contType},
		"Content-Length": {"0"},
		"x-amz-acl":      {string(perm)},
	}
	params := map[string][]string{
		"uploads": {""},
	}
	req := &request{
		method:  "POST",
		bucket:  b.Name,
		path:    key,
		headers: headers,
		params:  params,
	}
	var err error
	var resp struct {
		UploadID string `xml:"UploadId"`
	}
	for attempt := b.S3.AttemptStrategy.Start(); attempt.Next(); {
		err = b.S3.query(ctx, req, &resp)
		if !shouldRetry(err) {
			break
		}
	}
	if err != nil {
		return nil, err
	}
	return &Multi{Bucket: b, Key: key, UploadID: resp.UploadID}, nil
}
430106
func (d *dbBase) InsertMulti(q dbQuerier, mi *modelInfo, sind reflect.Value, bulk int, tz *time.Location) (int64, error) {
	var (
		cnt    int64
		nums   int
		values []interface{}
		names  []string
	)

	// typ := reflect.Indirect(mi.addrField).Type()

	length := sind.Len()

	for i := 1; i <= length; i++ {

		ind := reflect.Indirect(sind.Index(i - 1))

		// Is this needed ?
		// if !ind.Type().AssignableTo(typ) {
		// 	return cnt, ErrArgs
		// }

		if i == 1 {
			vus, err := d.collectValues(mi, ind, mi.fields.dbcols, true, true, &names, tz)
			if err != nil {
				return cnt, err
			}
			values = make([]interface{}, bulk*len(vus))
			nums += copy(values, vus)

		} else {

			vus, err := d.collectValues(mi, ind, mi.fields.dbcols, true, true, nil, tz)
			if err != nil {
				return cnt, err
			}

			if len(vus) != len(names) {
				return cnt, ErrArgs
			}

			nums += copy(values[nums:], vus)
		}

		if i > 1 && i%bulk == 0 || length == i {
			num, err := d.InsertValue(q, mi, true, names, values[:nums])
			if err != nil {
				return cnt, err
			}
			cnt += num
			nums = 0
		}
	}

	return cnt, nil
}
1022101
func (c *MockClient) GetMulti(keys []string) map[string]GetResponse {
	c.mutex.Lock()
	defer c.mutex.Unlock()

	res := make(map[string]GetResponse)
	for _, key := range keys {
		res[key] = c.getHelper(key)
	}
	return res
}
292070
func (g *Goon) DeleteMulti(keys []*datastore.Key) error {
	if len(keys) == 0 {
		return nil
		// not an error, and it was "successful", so return nil
	}
	memkeys := make([]string, len(keys))

	g.cacheLock.Lock()
	for i, k := range keys {
		mk := MemcacheKey(k)
		memkeys[i] = mk

		if g.inTransaction {
			g.txnCacheLock.Lock()
			g.toDelete[mk] = true
			g.txnCacheLock.Unlock()
		} else {
			delete(g.cache, mk)
		}
	}
	g.cacheLock.Unlock()

	// Memcache needs to be updated after the datastore to prevent a common race condition,
	// where a concurrent request will fetch the not-yet-updated data from the datastore
	// and populate memcache with it.
	if g.inTransaction {
		g.txnCacheLock.Lock()
		for _, mk := range memkeys {
			g.toDeleteMC[mk] = true
		}
		g.txnCacheLock.Unlock()
	} else {
		defer memcache.DeleteMulti(g.Context, memkeys)
	}

	mu := new(sync.Mutex)
	multiErr, any := make(appengine.MultiError, len(keys)), false
	goroutines := (len(keys)-1)/deleteMultiLimit + 1
	var wg sync.WaitGroup
	wg.Add(goroutines)
	for i := 0; i < goroutines; i++ {
		go func(i int) {
			defer wg.Done()
			lo := i * deleteMultiLimit
			hi := (i + 1) * deleteMultiLimit
			if hi > len(keys) {
				hi = len(keys)
			}
			dmerr := datastore.DeleteMulti(g.Context, keys[lo:hi])
			if dmerr != nil {
				mu.Lock()
				any = true // this flag tells DeleteMulti to return multiErr later
				mu.Unlock()
				merr, ok := dmerr.(appengine.MultiError)
				if !ok {
					g.error(dmerr)
					for j := lo; j < hi; j++ {
						multiErr[j] = dmerr
					}
					return
				}
				copy(multiErr[lo:hi], merr)
			}
		}(i)
	}
	wg.Wait()
	if any {
		return realError(multiErr)
	}
	return nil
}
326716
func PrepareBatchPayments(mctx libkb.MetaContext, walletState *WalletState, senderSeed stellarnet.SeedStr, payments []stellar1.BatchPaymentArg) ([]*MiniPrepared, func(), error) {
	mctx.Debug("preparing %d batch payments", len(payments))

	baseFee := walletState.BaseFee(mctx)

	prepared := make(chan *MiniPrepared)

	sp, unlock := NewSeqnoProvider(mctx, walletState)
	for _, payment := range payments {
		go func(p stellar1.BatchPaymentArg) {
			prepared <- prepareBatchPayment(mctx, walletState, sp, senderSeed, p, baseFee)
		}(payment)
	}

	// prepared chan could be out of order, so sort by seqno
	preparedList := make([]*MiniPrepared, len(payments))
	for i := 0; i < len(payments); i++ {
		preparedList[i] = <-prepared
	}
	sort.Slice(preparedList, func(a, b int) bool { return preparedList[a].Seqno < preparedList[b].Seqno })

	return preparedList, unlock, nil
}
704930
func (api *AutoBackupAPI) FilterMultiBy(key string, value interface{}) *AutoBackupAPI {
	api.filterBy(key, value, true)
	return api
}
861201
func (c *RawBinaryClient) AddMulti(items []*Item) []MutateResponse {
	return c.mutateMulti(opAdd, opAdd, items)
}