// NewBatch create a new batch to pack mutiple commands.
767965
func (cluster *Cluster) NewBatch() *Batch {
    return &Batch{
	cluster: cluster,
	batches: make([]nodeBatch, 0),
	index: make([]int, 0),
    }
}
608759
func NewBatch(typ BatchType) *Batch {
	return &Batch{
		Type:    typ,
		metrics: &queryMetrics{m: make(map[string]*hostMetrics)},
		spec:    &NonSpeculativeExecution{},
	}
}
220062
func NewBatchTaskState(stateName string,
	parameters BatchTaskParameters) *BatchTaskState {

	sns := &BatchTaskState{
		BaseTask: BaseTask{
			baseInnerState: baseInnerState{
				name: stateName,
				id:   rand.Int63(),
			},
		},
		parameters: parameters,
	}
	return sns
}
789139
func (s *Session) ExecuteBatch(batch *Batch) error {
	iter := s.executeBatch(batch)
	return iter.Close()
}
1253321
func NewBatchDeleter(ctx context.Context, wg *sync.WaitGroup, s *SQSService, every, drainTimeout time.Duration) chan<- *sqs.Message {
	dq := &deleteQueue{
		svc:                 s,
		accumulationTimeout: every,
		drainTimeout:        drainTimeout,
		queue:               make(chan *sqs.Message),
	}
	wg.Add(1)
	go dq.start(ctx, wg)
	return dq.queue
}
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
}
813125
func NewBatchMethodMetrics(
	scope tally.Scope,
	methodName string,
	samplingRate float64,
) BatchMethodMetrics {
	return BatchMethodMetrics{
		RetryableErrors:    scope.Counter(methodName + ".retryable-errors"),
		NonRetryableErrors: scope.Counter(methodName + ".non-retryable-errors"),
		Errors:             scope.Counter(methodName + ".errors"),
		Success:            scope.Counter(methodName + ".success"),
		Latency:            MustCreateSampledTimer(scope.Timer(methodName+".latency"), samplingRate),
	}
}
99031
func (s *LDBStore) writeBatch(b *dbBatch, wFlag uint8) error {
	if wFlag&wEntryCnt > 0 {
		b.Put(keyEntryCnt, U64ToBytes(s.entryCnt))
	}
	if wFlag&wIndexCnt > 0 {
		b.Put(keyDataIdx, U64ToBytes(s.dataIdx))
	}
	if wFlag&wAccessCnt > 0 {
		b.Put(keyAccessCnt, U64ToBytes(s.accessCnt))
	}
	l := b.Len()
	if err := s.db.Write(b.Batch); err != nil {
		return fmt.Errorf("unable to write batch: %v", err)
	}
	log.Trace(fmt.Sprintf("batch write (%d entries)", l))
	return nil
}
107532
func NewBatch(id []byte) *Batch {
	return &Batch{
		ID:          id,
		ReplaySet:   NewReplaySet(),
		entries:     make(map[uint16]batchEntry),
		replayCache: make(map[HashPrefix]struct{}),
	}
}
1187556
func (pdb *prefixDB) NewBatch() Batch {
	pdb.mtx.Lock()
	defer pdb.mtx.Unlock()

	return newPrefixBatch(pdb.prefix, pdb.db.NewBatch())
}
769485
func (s *SwarmSyncerServer) SetNextBatch(from, to uint64) ([]byte, uint64, uint64, *HandoverProof, error) {
	var batch []byte
	i := 0

	var ticker *time.Ticker
	defer func() {
		if ticker != nil {
			ticker.Stop()
		}
	}()
	var wait bool
	for {
		if wait {
			if ticker == nil {
				ticker = time.NewTicker(1000 * time.Millisecond)
			}
			select {
			case <-ticker.C:
			case <-s.quit:
				return nil, 0, 0, nil, nil
			}
		}

		metrics.GetOrRegisterCounter("syncer.setnextbatch.iterator", nil).Inc(1)
		err := s.store.Iterator(from, to, s.po, func(key storage.Address, idx uint64) bool {
			select {
			case <-s.quit:
				return false
			default:
			}
			batch = append(batch, key[:]...)
			i++
			to = idx
			return i < BatchSize
		})
		if err != nil {
			return nil, 0, 0, nil, err
		}
		if len(batch) > 0 {
			break
		}
		wait = true
	}

	log.Trace("Swarm syncer offer batch", "po", s.po, "len", i, "from", from, "to", to, "current store count", s.store.BinIndex(s.po))
	return batch, from, to, nil, nil
}
553698
func NewBatchServiceStateManager(facade Facade, ctx datastore.Context, runLevelTimeout time.Duration) *BatchServiceStateManager {
	return &BatchServiceStateManager{
		lock:             &sync.RWMutex{},
		currentStateLock: &sync.Mutex{},
		Facade:           facade,
		ctx:              ctx,
		ServiceRunLevelTimeout: runLevelTimeout,
		TenantQueues:           make(map[string]map[service.DesiredState]*ServiceStateQueue),
		TenantShutDowns:        make(map[string]chan<- int),
		currentStateWaits:      make(map[string]*CurrentStateWait),
	}
}
526046
func NewAutoBatching(d ds.Batching, size int) *Datastore {
	return &Datastore{
		child:            d,
		buffer:           make(map[ds.Key]op, size),
		maxBufferEntries: size,
	}
}
66036
func (p *ConnPool) BeginBatch() *Batch {
	c, err := p.Acquire()
	return &Batch{conn: c, connPool: p, err: err}
}
176824
func (s *Serializer) SerializeBatch(metrics []telegraf.Metric) ([]byte, error) {
	s.buf.Reset()
	for _, m := range metrics {
		_, err := s.Write(&s.buf, m)
		if err != nil {
			return nil, err
		}
	}
	out := make([]byte, s.buf.Len())
	copy(out, s.buf.Bytes())
	return out, nil
}
844806
func (m *collection) NewBatch(totalOps, totalKeyValBytes int) (
	Batch, error) {
	if m.isClosed() {
		return nil, ErrClosed
	}

	atomic.AddUint64(&m.stats.TotNewBatch, 1)
	atomic.AddUint64(&m.stats.TotNewBatchTotalOps, uint64(totalOps))
	atomic.AddUint64(&m.stats.TotNewBatchTotalKeyValBytes, uint64(totalKeyValBytes))

	return newBatch(m, BatchOptions{totalOps, totalKeyValBytes})
}
143878
func newSyncMemBatch() *syncMemBatch {
	return &syncMemBatch{
		batch: make(map[common.Hash][]byte),
		order: make([]common.Hash, 0, 256),
	}
}