// Batch is an internal RPC method used to process batch requests.
195674
func (JSONRPC2) Batch(arg BatchArg, replies *[]*json.RawMessage) (err error) {
	cli, srv := net.Pipe()
	defer cli.Close()
	go arg.srv.ServeCodec(NewServerCodecContext(arg.Context(), srv, arg.srv))

	replyc := make(chan *json.RawMessage, len(arg.reqs))
	donec := make(chan struct{}, 1)

	go func() {
		dec := json.NewDecoder(cli)
		*replies = make([]*json.RawMessage, 0, len(arg.reqs))
		for reply := range replyc {
			if reply != nil {
				*replies = append(*replies, reply)
			} else {
				*replies = append(*replies, new(json.RawMessage))
				if dec.Decode((*replies)[len(*replies)-1]) != nil {
					(*replies)[len(*replies)-1] = &jErrRequest
				}
			}
		}
		donec <- struct{}{}
	}()

	var testreq serverRequest
	for _, req := range arg.reqs {
		if req == nil || json.Unmarshal(*req, &testreq) != nil {
			replyc <- &jErrRequest
		} else {
			if testreq.ID != nil {
				replyc <- nil
			}
			if _, err = cli.Write(append(*req, '\n')); err != nil {
				break
			}
		}
	}

	close(replyc)
	<-donec
	return err
}
171638
func (b *Batch) Put(key, value []byte) {
	b.appendRec(keyTypeVal, key, value)
}
107532
func NewBatch(id []byte) *Batch {
	return &Batch{
		ID:          id,
		ReplaySet:   NewReplaySet(),
		entries:     make(map[uint16]batchEntry),
		replayCache: make(map[HashPrefix]struct{}),
	}
}
659867
func (h *Handler) flush(batch *batch.Batch) {
	size := batch.Size()
	if err := batch.Flush(); err != nil {
		stdlog.Printf("log/elastic: failed to flush %d logs: %s", size, err)
	}
}
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)
	}
}
695867
func leveldb_writebatch_iterate_put(p unsafe.Pointer, k *C.char, klen C.size_t, v *C.char, vlen C.size_t) {
	b := (*leveldb.Batch)(p)
	key := slice(unsafe.Pointer(k), int(klen))
	value := slice(unsafe.Pointer(v), int(vlen))
	b.Put(key, value)
}
354643
func (p *protoHandler) Echo(ctx context.Context, token *crossdockpb.Token) error {
	callBackAddr := yarpc.CallFromContext(ctx).Header(callBackAddrHeader)
	p.callHome(ctx, callBackAddr, []byte(token.Value), protobuf.Encoding)
	return nil
}
789139
func (s *Session) ExecuteBatch(batch *Batch) error {
	iter := s.executeBatch(batch)
	return iter.Close()
}
1121421
func (d *Failstore) Batch() (ds.Batch, error) {
	if err := d.errfunc("batch"); err != nil {
		return nil, err
	}

	b, err := d.child.(ds.Batching).Batch()
	if err != nil {
		return nil, err
	}

	return &FailBatch{
		cb:     b,
		dstore: d,
	}, nil
}
105821
func (b *Batcher) Wait() int {
	numWaiters := b.waiters.Add(1)
	if numWaiters == 1 {
		b.newBatch()
	}
	return <-b.queue
}
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
}
377127
func leveldb_writebatch_iterate_delete(p unsafe.Pointer, k *C.char, klen C.size_t) {
	b := (*leveldb.Batch)(p)
	key := slice(unsafe.Pointer(k), int(klen))
	b.Delete(key)
}
1067103
func generateBatchRequest(tokens []string, topic string) ([]byte, error) {
	envelope := new(BatchRequest)
	envelope.To = topics + extractTopicName(topic)
	envelope.RegTokens = make([]string, len(tokens))
	copy(envelope.RegTokens, tokens)

	return json.Marshal(envelope)

}
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()
}
1185335
func (b *batch) Write() error {
	return b.db.Write(b.b, nil)
}
708664
func (b *Batch) Queue(query string, arguments []interface{}, parameterOIDs []pgtype.OID, resultFormatCodes []int16) {
	b.items = append(b.items, &batchItem{
		query:             query,
		arguments:         arguments,
		parameterOIDs:     parameterOIDs,
		resultFormatCodes: resultFormatCodes,
	})
}
504262
func (hook *InfluxDBHook) handleBatch() {
	if hook.batchInterval == 0 || hook.batchCount == 0 {
		// we don't need to process this if the interval is 0
		return
	}
	for {
		time.Sleep(hook.batchInterval)
		hook.Lock()
		hook.writePoints()
		hook.Unlock()
	}
}