// this function takes data from socketRead and calls processBatch on a batch of it at a time
// then the resulting bytes are written to wocketWrite as fast as possible
177501
func ProcessBatchedData(socketRead io.ReadCloser, socketWrite io.Writer,
	makeProcessBatch func() (func([]byte) []byte, func([]byte, []byte)),
	batchSize int, workItemSize int) {
	readChan := make(chan []byte, 2)
	writeChan := make(chan []byte, 2)
	go readBatch(readChan, socketRead, batchSize, workItemSize)
	go writeBatch(writeChan, socketWrite)
	processBatch, prefetchBatch := makeProcessBatch()
	for buf := range readChan {
		res := processBatch(buf)
		writeChan <- res
		prefetchBatch(buf, res)
	}
	close(writeChan)
}
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)
}
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
}
235535
func readCallback(a *C.utp_callback_arguments) C.uint64 {
	s := libContextToSocket[a.context]
	c := s.conns[a.socket]
	b := a.bufBytes()
	if logCallbacks {
		log.Printf("read callback: conn %p: %d bytes", c, len(b))
	}
	if len(b) == 0 {
		panic("that will break the read drain invariant")
	}
	c.readBuf.Write(b)
	c.cond.Broadcast()
	return 0
}
1008789
func (b *batchProducer) Flush(timeout time.Duration, sendStats bool) (int, int, error) {
	b.Stop()

	timer := time.NewTimer(timeout)
	if timeout == 0 {
		timer.Stop()
	}

	timedOut := false
	sent := 0

loop:
	for len(b.records) > 0 {
		select {
		case <-timer.C:
			timedOut = true
			break loop
		default:
			sent += b.sendBatch(MaxKinesisBatchSize)
		}
	}

	if !timedOut && sendStats {
		b.sendStats()
	}

	return sent, len(b.records), nil
}
1048037
func CopyBatch(dst BatchWriter, src Reader, batchSize int) (cnt int, err error) {
	if batchSize <= 0 {
		batchSize = DefaultBatch
	}
	buf := make([]Quad, batchSize)
	bsrc, ok := src.(BatchReader)
	if !ok {
		bsrc = batchReader{src}
	}
	var n int
	for err == nil {
		n, err = bsrc.ReadQuads(buf)
		if err != nil && err != io.EOF {
			return
		}
		eof := err == io.EOF
		n, err = dst.WriteQuads(buf[:n])
		cnt += n
		if eof {
			break
		}
	}
	return
}
1075760
func (b *Batcher) write(ctx context.Context, org, bucket platform.ID, lines <-chan []byte, errC chan<- error) {
	flushInterval := b.MaxFlushInterval
	if flushInterval == 0 {
		flushInterval = DefaultInterval
	}

	maxBytes := b.MaxFlushBytes
	if maxBytes == 0 {
		maxBytes = DefaultMaxBytes
	}

	timer := time.NewTimer(flushInterval)
	defer func() { _ = timer.Stop() }()

	buf := make([]byte, 0, maxBytes)
	r := bytes.NewReader(buf)

	var line []byte
	var more = true
	// if read closes the channel normally, exit the loop
	for more {
		select {
		case line, more = <-lines:
			if more {
				buf = append(buf, line...)
			}
			// write if we exceed the max lines OR read routine has finished
			if len(buf) >= maxBytes || (!more && len(buf) > 0) {
				r.Reset(buf)
				timer.Reset(flushInterval)
				if err := b.Service.Write(ctx, org, bucket, r); err != nil {
					errC <- err
					return
				}
				buf = buf[:0]
			}
		case <-timer.C:
			if len(buf) > 0 {
				r.Reset(buf)
				timer.Reset(flushInterval)
				if err := b.Service.Write(ctx, org, bucket, r); err != nil {
					errC <- err
					return
				}
				buf = buf[:0]
			}
		case <-ctx.Done():
			errC <- ctx.Err()
			return
		}
	}

	errC <- nil
}
893316
func (c *RawKVClient) BatchPut(keys, values [][]byte) error {
	start := time.Now()
	defer func() {
		tikvRawkvCmdHistogramWithBatchPut.Observe(time.Since(start).Seconds())
	}()

	if len(keys) != len(values) {
		return errors.New("the len of keys is not equal to the len of values")
	}
	for _, value := range values {
		if len(value) == 0 {
			return errors.New("empty value is not supported")
		}
	}
	bo := NewBackoffer(context.Background(), rawkvMaxBackoff)
	err := c.sendBatchPut(bo, keys, values)
	return errors.Trace(err)
}
789139
func (s *Session) ExecuteBatch(batch *Batch) error {
	iter := s.executeBatch(batch)
	return iter.Close()
}
1094614
func readUntilNullWorkSizeBatch(socketRead io.ReadCloser,
	batch []byte, workSize int) (size int, err error) {
	err = nil
	size = 0
	if workSize == 0 {
		size, err = io.ReadFull(socketRead, batch)
	} else {
		lastCheckedForNull := 0
		for err == nil {
			var offset int
			offset, err = socketRead.Read(batch[size:])
			size += offset
			if err == nil && size < len(batch) {
				endCheck := size - (size % workSize)
				if zeroWorkItem(batch[lastCheckedForNull:endCheck], workSize) {
					if size%workSize != 0 {
						rem := workSize - (size % workSize)
						offset, err = io.ReadFull(socketRead, batch[size:size+rem])
						size += offset
					}
					return
				}
				lastCheckedForNull = endCheck // need to check partial work items
			} else {
				return
			}
		}
	}
	return
}
1149434
func (c *CircularBuffer) PushBatch(events []Event) {
	c.Lock()
	defer c.Unlock()

	for i := range events {
		c.push(events[i])
	}
}
1101048
func (l *Ledis) newBatch(wb *store.WriteBatch, locker sync.Locker) *batch {
	b := new(batch)
	b.l = l
	b.WriteBatch = wb

	b.Locker = locker

	return b
}
1044934
func (mBatch *cLevelDBBatch) Write() {
	err := mBatch.db.db.Write(mBatch.db.wo, mBatch.batch)
	if err != nil {
		panic(err)
	}
}
1154588
func executeBatches(batches []batch) error {
	logger.Debugf("Executing batches = %s", batches)
	numBatches := len(batches)
	if numBatches == 0 {
		return nil
	}
	if numBatches == 1 {
		return batches[0].execute()
	}
	var batchWG sync.WaitGroup
	batchWG.Add(numBatches)
	errsChan := make(chan error, numBatches)
	defer close(errsChan)
	for _, b := range batches {
		go func(b batch) {
			defer batchWG.Done()
			if err := b.execute(); err != nil {
				errsChan <- err
				return
			}
		}(b)
	}
	batchWG.Wait()
	if len(errsChan) > 0 {
		return <-errsChan
	}
	return nil
}
645670
func (b *bufferedBackend) processEvents(events []*auditinternal.Event) {
	if len(events) == 0 {
		return
	}

	// TODO(audit): Should control the number of active goroutines
	// if one goroutine takes 5 seconds to finish, the number of goroutines can be 5 * defaultBatchThrottleQPS
	if b.throttle != nil {
		b.throttle.Accept()
	}

	if b.asyncDelegate {
		b.wg.Add(1)
		go func() {
			defer b.wg.Done()
			defer runtime.HandleCrash()

			// Execute the real processing in a goroutine to keep it from blocking.
			// This lets the batching routine continue draining the queue immediately.
			b.delegateBackend.ProcessEvents(events...)
		}()
	} else {
		func() {
			defer runtime.HandleCrash()

			// Execute the real processing in a goroutine to keep it from blocking.
			// This lets the batching routine continue draining the queue immediately.
			b.delegateBackend.ProcessEvents(events...)
		}()
	}
}
1155746
func batchRawPut(value []byte) {
	cli, err := tikv.NewRawKVClient(strings.Split(*pdAddr, ","), config.Security{
		ClusterSSLCA:   *sslCA,
		ClusterSSLCert: *sslCert,
		ClusterSSLKey:  *sslKey,
	})
	if err != nil {
		log.Fatal(err.Error())
	}

	wg := sync.WaitGroup{}
	base := *dataCnt / *workerCnt
	wg.Add(*workerCnt)
	for i := 0; i < *workerCnt; i++ {
		go func(i int) {
			defer wg.Done()

			for j := 0; j < base; j++ {
				k := base*i + j
				key := fmt.Sprintf("key_%d", k)
				err = cli.Put([]byte(key), value)
				if err != nil {
					log.Fatal("put failed", zap.Error(err))
				}
			}
		}(i)
	}
	wg.Wait()
}
143878
func newSyncMemBatch() *syncMemBatch {
	return &syncMemBatch{
		batch: make(map[common.Hash][]byte),
		order: make([]common.Hash, 0, 256),
	}
}