// RunBatch execute commands in batch simutaneously. If multiple commands are 
// directed to the same node, they will be merged and sent at once using pipeling.
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
}
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
}
1255269
func (f Uint64Field) PutInBatch(batch *leveldb.Batch, val uint64) {
	batch.Put(f.key, encodeUint64(val))
}
92280
func (prod *InfluxDB) sendBatch() core.AssemblyFunc {
	if prod.writer.isConnectionUp() {
		return prod.assembly.Write
	} else if prod.IsStopping() {
		return prod.assembly.Flush
	}

	return nil
}
608759
func NewBatch(typ BatchType) *Batch {
	return &Batch{
		Type:    typ,
		metrics: &queryMetrics{m: make(map[string]*hostMetrics)},
		spec:    &NonSpeculativeExecution{},
	}
}
1295726
func (c *trackBranchCommand) Run(ctx *cmd.Context) error {
	client, err := c.getAPI()
	if err != nil {
		return err
	}
	defer func() { _ = client.Close() }()

	return errors.Trace(client.TrackBranch(c.branchName, c.entities))
}
648664
func (m *MockBuildsServer) Batch(arg0 context.Context, arg1 *BatchRequest) (*BatchResponse, error) {
	m.ctrl.T.Helper()
	ret := m.ctrl.Call(m, "Batch", arg0, arg1)
	ret0, _ := ret[0].(*BatchResponse)
	ret1, _ := ret[1].(error)
	return ret0, ret1
}
314078
func (b *Batcher) Run(param interface{}) error {
	if b.prefilter != nil {
		if err := b.prefilter(param); err != nil {
			return err
		}
	}

	if b.timeout == 0 {
		return b.doWork([]interface{}{param})
	}

	w := &work{
		param:  param,
		future: make(chan error, 1),
	}

	b.submitWork(w)

	return <-w.future
}
1114235
func (db *MemDB) NewBatch() Batch {
	db.mtx.Lock()
	defer db.mtx.Unlock()

	return &memBatch{db, nil}
}
232036
func (q *query) ExecuteBatch(ctx context.Context, request *querypb.ExecuteBatchRequest) (response *querypb.ExecuteBatchResponse, err error) {
	defer q.server.HandlePanic(&err)
	ctx = callerid.NewContext(callinfo.GRPCCallInfo(ctx),
		request.EffectiveCallerId,
		request.ImmediateCallerId,
	)
	results, err := q.server.ExecuteBatch(ctx, request.Target, request.Queries, request.AsTransaction, request.TransactionId, request.Options)
	if err != nil {
		return nil, vterrors.ToGRPC(err)
	}
	return &querypb.ExecuteBatchResponse{
		Results: sqltypes.ResultsToProto3(results),
	}, nil
}
133174
func (c *HoneycombClient) SendBatch(batch Batch) error {
	if len(batch) == 0 {
		// Nothing to send
		return nil
	}

	errs := []string{}
	for i := 0; i < len(batch); i += maxBatchSize {
		offset := i + maxBatchSize
		if offset > len(batch) {
			offset = len(batch)
		}
		if err := c.sendBatch(batch[i:offset]); err != nil {
			errs = append(errs, err.Error())
		}
	}

	if len(errs) > 0 {
		return errors.New(strings.Join(errs, "\n"))
	}

	return nil
}
665241
func (db *Database) NewBatch() ethdb.Batch {
	return &batch{
		db: db.db,
		b:  new(leveldb.Batch),
	}
}
767965
func (cluster *Cluster) NewBatch() *Batch {
    return &Batch{
	cluster: cluster,
	batches: make([]nodeBatch, 0),
	index: make([]int, 0),
    }
}
1116916
func (a *AgentClientUDP) EmitBatch(batch *jaeger.Batch) error {
	a.thriftBuffer.Reset()
	a.client.SeqId = 0 // we have no need for distinct SeqIds for our one-way UDP messages
	if err := a.client.EmitBatch(batch); err != nil {
		return err
	}
	if a.thriftBuffer.Len() > a.maxPacketSize {
		return fmt.Errorf("Data does not fit within one UDP packet; size %d, max %d, spans %d",
			a.thriftBuffer.Len(), a.maxPacketSize, len(batch.Spans))
	}
	_, err := a.connUDP.Write(a.thriftBuffer.Bytes())
	return err
}
247384
func (b *batch) run() {
	b.db.batchMu.Lock()
	b.timer.Stop()
	// Make sure no new work is added to this batch, but don't break
	// other batches.
	if b.db.batch == b {
		b.db.batch = nil
	}
	b.db.batchMu.Unlock()

retry:
	for len(b.calls) > 0 {
		var failIdx = -1
		err := b.db.Update(func(tx *Tx) error {
			for i, c := range b.calls {
				if err := safelyCall(c.fn, tx); err != nil {
					failIdx = i
					return err
				}
			}
			return nil
		})

		if failIdx >= 0 {
			// take the failing transaction out of the batch. it's
			// safe to shorten b.calls here because db.batch no longer
			// points to us, and we hold the mutex anyway.
			c := b.calls[failIdx]
			b.calls[failIdx], b.calls = b.calls[len(b.calls)-1], b.calls[:len(b.calls)-1]
			// tell the submitter re-run it solo, continue with the rest of the batch
			c.err <- trySolo
			continue retry
		}

		// pass success, or bolt internal errors, to all callers
		for _, c := range b.calls {
			c.err <- err
		}
		break retry
	}
}
791423
func (m *MockCommandRunner) RunCommands(arg0 exec.RunParams) (*exec.ExecResponse, error) {
	ret := m.ctrl.Call(m, "RunCommands", arg0)
	ret0, _ := ret[0].(*exec.ExecResponse)
	ret1, _ := ret[1].(error)
	return ret0, ret1
}
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
}