// ExecuteBatch atomically incorporates the provided Batch into the
// collection.  The Batch instance should not be reused after
// ExecuteBatch() returns.
420050
func (m *collection) ExecuteBatch(bIn Batch,
	writeOptions WriteOptions) error {
	startTime := time.Now()

	defer func() {
		m.fireEvent(EventKindBatchExecute, time.Now().Sub(startTime))
	}()

	atomic.AddUint64(&m.stats.TotExecuteBatchBeg, 1)

	b, ok := bIn.(*batch)
	if !ok {
		atomic.AddUint64(&m.stats.TotExecuteBatchErr, 1)

		return fmt.Errorf("wrong Batch implementation type")
	}

	if b == nil || b.isEmpty() {
		atomic.AddUint64(&m.stats.TotExecuteBatchEmpty, 1)

		m.histograms["ExecuteBatchUsecs"].Add(
			uint64(time.Since(startTime).Nanoseconds()/1000), 1)

		return nil
	}

	maxPreMergerBatches := m.options.MaxPreMergerBatches
	if maxPreMergerBatches <= 0 {
		maxPreMergerBatches =
			DefaultCollectionOptions.MaxPreMergerBatches
	}

	if m.options.DeferredSort {
		b.readyDeferredSort() // Recursively ready child batches.
	} else {
		b.doSort() // Recursively sort the child batches.
	}

	// Notify handlers that we are about to execute a batch.
	m.fireEvent(EventKindBatchExecuteStart, 0)

	m.m.Lock()

	for m.stackDirtyTop != nil &&
		len(m.stackDirtyTop.a) >= maxPreMergerBatches {
		if m.isClosed() {
			m.m.Unlock()
			return ErrClosed
		}

		if m.options.DeferredSort {
			go b.RequestSort() // While waiting, might as well sort.
		}

		atomic.AddUint64(&m.stats.TotExecuteBatchWaitBeg, 1)
		m.stackDirtyTopCond.Wait()
		atomic.AddUint64(&m.stats.TotExecuteBatchWaitEnd, 1)
	}

	if m.isClosed() { // Could have been closed while waiting.
		m.m.Unlock()
		return ErrClosed
	}

	m.invalidateLatestSnapshotLOCKED()

	stackDirtyTop := m.buildStackDirtyTop(b, m.stackDirtyTop)

	prevStackDirtyTop := m.stackDirtyTop
	m.stackDirtyTop = stackDirtyTop

	waitDirtyIncomingCh := m.waitDirtyIncomingCh
	m.waitDirtyIncomingCh = nil

	m.m.Unlock()

	prevStackDirtyTop.Close()

	if waitDirtyIncomingCh != nil {
		atomic.AddUint64(&m.stats.TotExecuteBatchAwakeMergerBeg, 1)
		close(waitDirtyIncomingCh)
		atomic.AddUint64(&m.stats.TotExecuteBatchAwakeMergerEnd, 1)
	}

	atomic.AddUint64(&m.stats.TotExecuteBatchEnd, 1)

	m.histograms["ExecuteBatchUsecs"].Add(
		uint64(time.Since(startTime).Nanoseconds()/1000), 1)

	return nil
}
95715
func (q *lockBasedQueryExecutor) ExecuteQuery(namespace, query string) (commonledger.ResultsIterator, error) {
	return q.helper.executeQuery(namespace, query)
}
1043165
func deleteBatch(ctx aws.Context, d *BatchDelete, input *s3.DeleteObjectsInput, objects []BatchDeleteObject) []Error {
	errs := []Error{}

	if result, err := d.Client.DeleteObjectsWithContext(ctx, input); err != nil {
		for i := 0; i < len(input.Delete.Objects); i++ {
			errs = append(errs, newError(err, input.Bucket, input.Delete.Objects[i].Key))
		}
	} else if len(result.Errors) > 0 {
		for i := 0; i < len(result.Errors); i++ {
			code := ErrDeleteBatchFailCode
			msg := errDefaultDeleteBatchMessage
			if result.Errors[i].Message != nil {
				msg = *result.Errors[i].Message
			}
			if result.Errors[i].Code != nil {
				code = *result.Errors[i].Code
			}

			errs = append(errs, newError(awserr.New(code, msg, err), input.Bucket, result.Errors[i].Key))
		}
	}
	for _, object := range objects {
		if object.After == nil {
			continue
		}
		if err := object.After(); err != nil {
			errs = append(errs, newError(err, object.Object.Bucket, object.Object.Key))
		}
	}

	return errs
}
353425
func (tx *Tx) BeginBatch() *Batch {
	return &Batch{conn: tx.conn, inTx: true}
}
1270454
func (dbInst *DB) WriteBatch(batch *leveldb.Batch, sync bool) error {
	wo := dbInst.writeOptsNoSync
	if sync {
		wo = dbInst.writeOptsSync
	}
	if err := dbInst.db.Write(batch, wo); err != nil {
		return errors.Wrap(err, "error writing batch to leveldb")
	}
	return nil
}
450534
func (s *Session) NewBatch(typ BatchType) *Batch {
	s.mu.RLock()
	batch := &Batch{
		Type:             typ,
		rt:               s.cfg.RetryPolicy,
		serialCons:       s.cfg.SerialConsistency,
		observer:         s.batchObserver,
		Cons:             s.cons,
		defaultTimestamp: s.cfg.DefaultTimestamp,
		keyspace:         s.cfg.Keyspace,
		metrics:          &queryMetrics{m: make(map[string]*hostMetrics)},
		spec:             &NonSpeculativeExecution{},
	}

	s.mu.RUnlock()
	return batch
}
851227
func (job *RescanJob) batch() *rescanBatch {
	return &rescanBatch{
		initialSync: job.InitialSync,
		addrs:       job.Addrs,
		outpoints:   job.OutPoints,
		bs:          job.BlockStamp,
		errChans:    []chan error{job.err},
	}
}
937037
func NewBatch(n int64) *Batch {
	return &Batch{
		Ops: make([]*types.TxResult, n),
	}
}
1149454
func (res *Resolver) ExecuteBatch(
	ctx context.Context,
	tabletType topodatapb.TabletType,
	asTransaction bool,
	session *vtgatepb.Session,
	options *querypb.ExecuteOptions,
	buildBatchRequest func() (*scatterBatchRequest, error),
) ([]sqltypes.Result, error) {
	batchRequest, err := buildBatchRequest()
	if err != nil {
		return nil, err
	}
	for {
		qrs, err := res.scatterConn.ExecuteBatch(
			ctx,
			batchRequest,
			tabletType,
			asTransaction,
			NewSafeSession(session),
			options)
		// Don't retry transactional requests.
		if asTransaction {
			return qrs, err
		}
		// If lower level retries failed, check if there was a resharding event
		// and retry again if needed.
		if isRetryableError(err) {
			newBatchRequest, buildErr := buildBatchRequest()
			if buildErr != nil {
				return nil, buildErr
			}
			// Use reflect to see if the request has changed.
			if reflect.DeepEqual(*batchRequest, *newBatchRequest) {
				return qrs, err
			}
			batchRequest = newBatchRequest
			continue
		}
		return qrs, err
	}
}
1055353
func (clnt *Client) batchExecute(policy *BatchPolicy, keys []*Key, cmd batcher) error {
	batchNodes, err := newBatchNodeList(clnt.cluster, policy, keys)
	if err != nil {
		return err
	}

	var wg sync.WaitGroup

	// Use a goroutine per namespace per node
	errs := []error{}
	errm := new(sync.Mutex)

	for _, batchNode := range batchNodes {
		// There may be multiple goroutines for a single node because the
		// wire protocol only allows one namespace per command.  Multiple namespaces
		// require multiple goroutines per node.
		batchNode.splitByNamespace(keys)

		wg.Add(len(batchNode.BatchNamespaces))
	}

	for _, batchNode := range batchNodes {
		for _, bns := range batchNode.BatchNamespaces {
			newCmd := cmd.cloneBatchCommand(batchNode, bns)
			go func(cmd command) {
				defer wg.Done()
				if err := cmd.Execute(); err != nil {
					errm.Lock()
					errs = append(errs, err)
					errm.Unlock()
				}
			}(newCmd)
		}
	}

	wg.Wait()
	return mergeErrors(errs)
}
547225
func (cluster *Cluster) RunBatch(bat *Batch) ([]interface{}, error) {
    for i := range bat.batches {
	go doBatch(&bat.batches[i])
    }

    for i := range bat.batches {
	<-bat.batches[i].done
    }

    var replies []interface{}
    for _, i := range bat.index {
	if bat.batches[i].err != nil {
	    return nil, bat.batches[i].err
	}

	replies = append(replies, bat.batches[i].cmds[0].reply)
	bat.batches[i].cmds = bat.batches[i].cmds[1:]
    }

    return replies, nil
}
1086704
func (conn *FakeVTGateConn) ExecuteBatchShards(ctx context.Context, queries []*vtgatepb.BoundShardQuery, tabletType topodatapb.TabletType, asTransaction bool, session *vtgatepb.Session, options *querypb.ExecuteOptions) (*vtgatepb.Session, []sqltypes.Result, error) {
	panic("not implemented")
}
268497
func (t *table) NewBatch() ethdb.Batch {
	return &tableBatch{t.db.NewBatch(), t.prefix}
}
665241
func (db *Database) NewBatch() ethdb.Batch {
	return &batch{
		db: db.db,
		b:  new(leveldb.Batch),
	}
}
589480
func (sbc *SandboxConn) ExecuteBatch(ctx context.Context, target *querypb.Target, queries []*querypb.BoundQuery, asTransaction bool, transactionID int64, options *querypb.ExecuteOptions) ([]sqltypes.Result, error) {
	sbc.ExecCount.Add(1)
	if asTransaction {
		sbc.AsTransactionCount.Add(1)
	}
	if err := sbc.getError(); err != nil {
		return nil, err
	}
	sbc.BatchQueries = append(sbc.BatchQueries, queries)
	sbc.Options = append(sbc.Options, options)
	result := make([]sqltypes.Result, 0, len(queries))
	for range queries {
		result = append(result, *(sbc.getNextResult()))
	}
	return result, 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
}
955156
func (queue *redisQueue) consumeBatch(batchSize int) bool {
	if batchSize == 0 {
		return false
	}

	for i := 0; i < batchSize; i++ {
		value, ok := queue.redisClient.RPopLPush(queue.readyKey, queue.unackedKey)
		if !ok {
			// debug(fmt.Sprintf("rmq queue consumed last batch %s %d", queue, i)) // COMMENTOUT
			return false
		}

		// debug(fmt.Sprintf("consume %d/%d %s %s", i, batchSize, value, queue)) // COMMENTOUT
		queue.deliveryChan <- newDelivery(value, queue.unackedKey, queue.rejectedKey, queue.pushKey, queue.redisClient)
	}

	// debug(fmt.Sprintf("rmq queue consumed batch %s %d", queue, batchSize)) // COMMENTOUT
	return true
}