// NewBatch creates new RPC batch for the given DispatcherClient.
//
// It is safe creating multiple concurrent batches from a single client.
434363
func (dc *DispatcherClient) NewBatch() *DispatcherBatch {
	return &DispatcherBatch{
		c: dc,
		b: dc.c.NewBatch(),
	}
}
920728
func NewBatchPolicy() *BatchPolicy {
	return &BatchPolicy{
		BasePolicy:          *NewPolicy(),
		ConcurrentNodes:     1,
		AllowInline:         true,
		AllowPartialResults: false,
		SendSetName:         false,
	}
}
1143420
func (OpenTracingTracer) TraceBatch(ctx context.Context, keys Keys) (context.Context, TraceBatchFinishFunc) {
	span, spanCtx := opentracing.StartSpanFromContext(ctx, "Dataloader: batch")

	span.SetTag("dataloader.keys", keys.Keys())

	return spanCtx, func(results []*Result) {
		// TODO: is there anything we should do with the results?
		span.Finish()
	}
}
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
}
353425
func (tx *Tx) BeginBatch() *Batch {
	return &Batch{conn: tx.conn, inTx: true}
}
608759
func NewBatch(typ BatchType) *Batch {
	return &Batch{
		Type:    typ,
		metrics: &queryMetrics{m: make(map[string]*hostMetrics)},
		spec:    &NonSpeculativeExecution{},
	}
}
1248589
func (stc *ScatterConn) ExecuteBatch(
	ctx context.Context,
	batchRequest *scatterBatchRequest,
	tabletType topodatapb.TabletType,
	asTransaction bool,
	session *SafeSession,
	options *querypb.ExecuteOptions) (qrs []sqltypes.Result, err error) {

	allErrors := new(concurrency.AllErrorRecorder)

	results := make([]sqltypes.Result, batchRequest.length)
	var resMutex sync.Mutex

	var wg sync.WaitGroup
	for _, req := range batchRequest.requests {
		wg.Add(1)
		go func(req *shardBatchRequest) {
			defer wg.Done()
			var err error
			startTime, statsKey := stc.startAction("ExecuteBatch", req.rs.Target)
			defer stc.endAction(startTime, allErrors, statsKey, &err, session)

			shouldBegin, transactionID := transactionInfo(req.rs.Target, session, false)
			var innerqrs []sqltypes.Result
			if shouldBegin {
				innerqrs, transactionID, err = req.rs.QueryService.BeginExecuteBatch(ctx, req.rs.Target, req.queries, asTransaction, options)
				if transactionID != 0 {
					if appendErr := session.Append(&vtgatepb.Session_ShardSession{
						Target:        req.rs.Target,
						TransactionId: transactionID,
					}, stc.txConn.mode); appendErr != nil {
						err = appendErr
					}
				}
				if err != nil {
					return
				}
			} else {
				innerqrs, err = req.rs.QueryService.ExecuteBatch(ctx, req.rs.Target, req.queries, asTransaction, transactionID, options)
				if err != nil {
					return
				}
			}

			resMutex.Lock()
			defer resMutex.Unlock()
			for i, result := range innerqrs {
				results[req.resultIndexes[i]].AppendResult(&result)
			}
		}(req)
	}
	wg.Wait()

	if session.MustRollback() {
		stc.txConn.Rollback(ctx, session)
	}
	if allErrors.HasErrors() {
		return nil, allErrors.AggrError(vterrors.Aggregate)
	}
	return results, nil
}
688058
func NewBatcher(interval time.Duration) *Batcher {
	return &Batcher{
		interval: interval,
		queue:    make(chan int),
		waiters:  NewAtomicInt32(0),
		nextID:   NewAtomicInt32(0),
		after:    time.After,
	}
}
14251
func NewDeadPingBatcher(err error) *PingBatcher {
	// we never start the loop, so the timeout doesn't matter.
	pb := &PingBatcher{}
	pb.tomb.Kill(err)
	return pb
}
529987
func SkipBatch() func(Call) error {
	return func(c Call) error {
		if b, ok := c.(Batchable); ok {
			b.setSkipBatch(true)
			return nil
		}
		return errors.New("'SkipBatch' option only works with Get and Mutate requests")
	}
}
1114235
func (db *MemDB) NewBatch() Batch {
	db.mtx.Lock()
	defer db.mtx.Unlock()

	return &memBatch{db, nil}
}
1191576
func (ddb debugDB) NewBatch() Batch {
	fmt.Printf("%v.NewBatch()\n", ddb.label)
	return NewDebugBatch(ddb.label, ddb.db.NewBatch())
}
665241
func (db *Database) NewBatch() ethdb.Batch {
	return &batch{
		db: db.db,
		b:  new(leveldb.Batch),
	}
}
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
}
717593
func NewCustomBatchProvider(ipt BaseIPTables, commit func(buf *bytes.Buffer) error, batchTables []string) *BatchProvider {

	batchTablesMap := map[string]bool{}

	for _, t := range batchTables {
		batchTablesMap[t] = true
	}

	return &BatchProvider{
		ipt:         ipt,
		rules:       map[string]map[string][]string{},
		batchTables: batchTablesMap,
		commitFunc:  commit,
	}
}
1024543
func (t *explainTablet) ExecuteBatch(ctx context.Context, target *querypb.Target, queries []*querypb.BoundQuery, asTransaction bool, transactionID int64, options *querypb.ExecuteOptions) ([]sqltypes.Result, error) {
	t.mu.Lock()
	t.currentTime = batchTime.Wait()

	// Since the query is simulated being "sent" over the wire we need to
	// copy the bindVars into the executor to avoid a data race.
	for _, query := range queries {
		query.BindVariables = sqltypes.CopyBindVariables(query.BindVariables)
		t.tabletQueries = append(t.tabletQueries, &TabletQuery{
			Time:     t.currentTime,
			SQL:      query.Sql,
			BindVars: query.BindVariables,
		})
	}
	t.mu.Unlock()

	return t.tsv.ExecuteBatch(ctx, target, queries, asTransaction, transactionID, options)
}
526046
func NewAutoBatching(d ds.Batching, size int) *Datastore {
	return &Datastore{
		child:            d,
		buffer:           make(map[ds.Key]op, size),
		maxBufferEntries: size,
	}
}