// Batched returns an implementation of the Handler interface that multiplexes
// requests on to a connection pool in order to reduce the overhead per request.
353064
func Batched(sock string, opts batched.Opts) handlers.HandlerConst {
	return func() (handlers.Handler, error) {
		return batched.NewHandler(sock, opts), nil
	}
}
430513
func NewShardedFlushingBatch(index ShardedIndex, maxBatchSize int) FlushingBatch {
	indices := index.shards()
	b := &shardedIndexFlushingBatch{
		batches: make([]*singleIndexFlushingBatch, len(indices)),
	}
	for i, index := range indices {
		b.batches[i] = newFlushingBatch(index, maxBatchSize)
	}
	return b
}
258061
func (dds *Delayed) Batch() (ds.Batch, error) {
	return ds.NewBasicBatch(dds), nil
}
666739
func (mvcc *MVCCLevelDB) BatchGet(ks [][]byte, startTS uint64, isoLevel kvrpcpb.IsolationLevel) []Pair {
	mvcc.mu.RLock()
	defer mvcc.mu.RUnlock()

	pairs := make([]Pair, 0, len(ks))
	for _, k := range ks {
		v, err := mvcc.getValue(k, startTS, isoLevel)
		if v == nil && err == nil {
			continue
		}
		pairs = append(pairs, Pair{
			Key:   k,
			Value: v,
			Err:   errors.Trace(err),
		})
	}
	return pairs
}
369830
func (b *batcher) batch(originalContext context.Context) {
	var (
		keys     = make(Keys, 0)
		reqs     = make([]*batchRequest, 0)
		items    = make([]*Result, 0)
		panicErr interface{}
	)

	for item := range b.input {
		keys = append(keys, item.key)
		reqs = append(reqs, item)
	}

	ctx, finish := b.tracer.TraceBatch(originalContext, keys)
	defer finish(items)

	func() {
		defer func() {
			if r := recover(); r != nil {
				panicErr = r
				if b.silent {
					return
				}
				const size = 64 << 10
				buf := make([]byte, size)
				buf = buf[:runtime.Stack(buf, false)]
				log.Printf("Dataloader: Panic received in batch function:: %v\n%s", panicErr, buf)
			}
		}()
		items = b.batchFn(ctx, keys)
	}()

	if panicErr != nil {
		for _, req := range reqs {
			req.channel <- &Result{Error: fmt.Errorf("Panic received in batch function: %v", panicErr)}
			close(req.channel)
		}
		return
	}

	if len(items) != len(keys) {
		err := &Result{Error: fmt.Errorf(`
			The batch function supplied did not return an array of responses
			the same length as the array of keys.

			Keys:
			%v

			Values:
			%v
		`, keys, items)}

		for _, req := range reqs {
			req.channel <- err
			close(req.channel)
		}

		return
	}

	for i, req := range reqs {
		req.channel <- items[i]
		close(req.channel)
	}
}
365591
func (h *Hub) Batch(f func(), shouldLock bool) {
	if pdebug.Enabled {
		g := pdebug.Marker("Batch %t", shouldLock)
		defer g.End()
	}
	if shouldLock {
		// lock during this operation
		h.mutex.Lock()
		defer h.mutex.Unlock()
	}

	// temporarily set isSync = true
	o := h.isSync
	h.isSync = true
	defer func() { h.isSync = o }()

	// ignore panics
	defer func() { recover() }()

	f()
}
66036
func (p *ConnPool) BeginBatch() *Batch {
	c, err := p.Acquire()
	return &Batch{conn: c, connPool: p, err: err}
}
1219238
func (h *DBHandle) WriteBatch(batch *UpdateBatch, sync bool) error {
	if len(batch.KVs) == 0 {
		return nil
	}
	levelBatch := &leveldb.Batch{}
	for k, v := range batch.KVs {
		key := constructLevelKey(h.dbName, []byte(k))
		if v == nil {
			levelBatch.Delete(key)
		} else {
			levelBatch.Put(key, v)
		}
	}
	if err := h.db.WriteBatch(levelBatch, sync); err != nil {
		return err
	}
	return nil
}
145762
func (h *Handler) flush(batch *batch.Batch) {
	size := batch.Size()
	start := time.Now()
	stdlog.Printf("log/elastic: flushing %d logs", size)

	if err := batch.Flush(); err != nil {
		stdlog.Printf("log/elastic: failed to flush %d logs: %s", size, err)
	}

	stdlog.Printf("log/elastic: flushed %d logs in %s", size, time.Since(start))
}
767818
func (b *Batch) Flush() (err error) {
	if b.Size() == 0 {
		return nil
	}

	buf, err := b.Bytes()
	if err != nil {
		return err
	}

	b.Docs = nil

	return b.Elastic.Bulk(buf)
}
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
}
373542
func (m *Mercury) BatchDelete(ctx context.Context, keys []string, opts ...clientv3.OpOption) (*clientv3.TxnResponse, error) {
	txn := m.cliv3.Txn(ctx)
	ops := []clientv3.Op{}
	for _, key := range keys {
		op := clientv3.OpDelete(m.parseKey(key), opts...)
		ops = append(ops, op)
	}
	return txn.Then(ops...).Commit()
}
795844
func (i *indexImpl) NewBatch() *Batch {
	return &Batch{
		index:    i,
		internal: index.NewBatch(),
	}
}
598953
func batchOps(done <-chan struct{}, ops <-chan map[string]interface{}) <-chan []interface{} {
	c := make(chan []interface{})

	go func() {
		defer close(c)
		// In a loop grab as many elements as you can before you would block (the default case)
		// Only place non-empty batches into the output channel.
		elements := make([]interface{}, 0)

		// Send the current list of elements as a batch, unless it's empty. Returns whether or not a batch was sent.
		sendElements := func() bool {
			if len(elements) == 0 {
				return false
			}
			select {
			case c <- elements:
				elements = make([]interface{}, 0)
			case <-done:
			}
			return true
		}

		for {
			select {
			case elem, channelOpen := <-ops:
				if !channelOpen {
					sendElements()
					return
				}
				elements = append(elements, elem)
			default:
				if !sendElements() {
					// stop from busy-wait eating all cpu.
					time.Sleep(1 * time.Millisecond)
				}
			}
		}
	}()
	return c
}
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
}
1205605
func NewBatchProcessErr(completed interface{}, err error) error {
	return &types.ErrBatchProcess{Goof: goof.WithFieldE(
		"completed", completed, "batch processing error", err)}
}
767965
func (cluster *Cluster) NewBatch() *Batch {
    return &Batch{
	cluster: cluster,
	batches: make([]nodeBatch, 0),
	index: make([]int, 0),
    }
}