// GetBatch retrieves the next batch of hashes from the dbstore
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
}
725923
func (c *CryptoSuite) GetHash(opts core.HashOpts) (h hash.Hash, err error) {
	return c.BCCSP.GetHash(opts)
}
418835
func (c *RawKVClient) BatchGet(keys [][]byte) ([][]byte, error) {
	start := time.Now()
	defer func() {
		tikvRawkvCmdHistogramWithBatchGet.Observe(time.Since(start).Seconds())
	}()

	bo := NewBackoffer(context.Background(), rawkvMaxBackoff)
	resp, err := c.sendBatchReq(bo, keys, tikvrpc.CmdRawBatchGet)
	if err != nil {
		return nil, errors.Trace(err)
	}

	cmdResp := resp.RawBatchGet
	if cmdResp == nil {
		return nil, errors.Trace(ErrBodyMissing)
	}

	keyToValue := make(map[string][]byte, len(keys))
	for _, pair := range cmdResp.Pairs {
		keyToValue[string(pair.Key)] = pair.Value
	}

	values := make([][]byte, len(keys))
	for i, key := range keys {
		values[i] = keyToValue[string(key)]
	}
	return values, nil
}
324676
func (p *BlockPrefetchClient) GetBlock(hash *chainhash.Hash) (*wire.MsgBlock, error) {
	msgBlock, _, err := p.GetBlockData(hash)
	return msgBlock, err
}
7360
func (db *DB) GetBestBlockHash() string {
	hash, err := db.RetrieveBestBlockHash()
	if err != nil {
		log.Errorf("RetrieveBestBlockHash failed: %v", err)
		return ""
	}
	return hash
}
510336
func (t *InjectedSnapshot) BatchGet(keys []Key) (map[string][]byte, error) {
	t.cfg.RLock()
	defer t.cfg.RUnlock()
	if t.cfg.getError != nil {
		return nil, t.cfg.getError
	}
	return t.Snapshot.BatchGet(keys)
}
1187556
func (pdb *prefixDB) NewBatch() Batch {
	pdb.mtx.Lock()
	defer pdb.mtx.Unlock()

	return newPrefixBatch(pdb.prefix, pdb.db.NewBatch())
}
208899
func (b *BtcWallet) GetBlock(blockHash *chainhash.Hash) (*wire.MsgBlock, error) {
	return b.chain.GetBlock(blockHash)
}
598604
func (txi *TxIndex) AddBatch(b *txindex.Batch) error {
	storeBatch := txi.store.NewBatch()
	defer storeBatch.Close()

	for _, result := range b.Ops {
		hash := result.Tx.Hash()

		// index tx by tags
		for _, tag := range result.Result.Tags {
			if txi.indexAllTags || cmn.StringInSlice(string(tag.Key), txi.tagsToIndex) {
				storeBatch.Set(keyForTag(tag, result), hash)
			}
		}

		// index tx by height
		if txi.indexAllTags || cmn.StringInSlice(types.TxHeightKey, txi.tagsToIndex) {
			storeBatch.Set(keyForHeight(result), hash)
		}

		// index tx by hash
		rawBytes, err := cdc.MarshalBinaryBare(result)
		if err != nil {
			return err
		}
		storeBatch.Set(hash, rawBytes)
	}

	storeBatch.Write()
	return nil
}
269124
func NewBatchChecker(backend ExecutorState, params Params, blockchain contexts.Blockchain, logger *logging.Logger,
	options ...ExecutionOption) BatchExecutor {

	return newExecutor("CheckCache", false, params, backend, blockchain, nil,
		logger.WithScope("NewBatchExecutor"), options...)
}
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
}
1249304
func (db *CLevelDB) NewBatch() Batch {
	batch := levigo.NewWriteBatch()
	return &cLevelDBBatch{db, batch}
}
709939
func (a *AuditBroker) GetHash(ctx context.Context, name string, input string) (string, error) {
	a.RLock()
	defer a.RUnlock()
	be, ok := a.backends[name]
	if !ok {
		return "", fmt.Errorf("unknown audit backend %q", name)
	}

	return be.backend.GetHash(ctx, input)
}
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
}
268497
func (t *table) NewBatch() ethdb.Batch {
	return &tableBatch{t.db.NewBatch(), t.prefix}
}
795844
func (i *indexImpl) NewBatch() *Batch {
	return &Batch{
		index:    i,
		internal: index.NewBatch(),
	}
}
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
}