// Implements Batch.
425630
func (mBatch *goLevelDBBatch) WriteSync() {
	err := mBatch.db.db.Write(mBatch.batch, &opt.WriteOptions{Sync: true})
	if err != nil {
		panic(err)
	}
}
730223
func (db *GoLevelDB) NewBatch() Batch {
	batch := new(leveldb.Batch)
	return &goLevelDBBatch{db, batch}
}
376985
func Batch(mctx libkb.MetaContext, walletState *WalletState, arg stellar1.BatchLocalArg) (res stellar1.BatchResultLocal, err error) {
	mctx = mctx.WithLogTag("BATCH=" + arg.BatchID)

	startTime := time.Now()
	res.StartTime = stellar1.ToTimeMs(startTime)
	defer func() {
		if res.EndTime == 0 {
			res.EndTime = stellar1.ToTimeMs(time.Now())
		}
	}()

	// look up sender account
	senderAccountID, senderSeed, err := LookupSenderSeed(mctx)
	if err != nil {
		return res, err
	}

	mctx.Debug("Batch sender account ID: %s", senderAccountID)
	mctx.Debug("Batch size: %d", len(arg.Payments))

	// prepare the payments
	prepared, unlock, err := PrepareBatchPayments(mctx, walletState, senderSeed, arg.Payments)
	if err != nil {
		walletState.SeqnoUnlock()
		return res, err
	}

	res.PreparedTime = stellar1.ToTimeMs(time.Now())

	// make a listener that will get payment status updates
	listenerID, listenerCh, err := DefaultLoader(mctx.G()).GetListener()
	if err != nil {
		unlock()
		return res, err
	}
	defer DefaultLoader(mctx.G()).RemoveListener(listenerID)

	resultList := make([]stellar1.BatchPaymentResult, len(prepared))
	waiting := make(map[stellar1.TransactionID]int)

	// submit the payments
	// need to submit tx one at a time, in order
	for i := 0; i < len(prepared); i++ {
		if prepared[i] == nil {
			unlock()
			// this should never happen
			return res, errors.New("batch prepare failed")
		}

		bpResult := stellar1.BatchPaymentResult{
			Username:  prepared[i].Username.String(),
			StartTime: stellar1.ToTimeMs(time.Now()),
		}
		if prepared[i].Error != nil {
			makeResultError(&bpResult, prepared[i].Error)
		} else {
			submitBatchTx(mctx, walletState, senderAccountID, prepared[i], &bpResult)
			if bpResult.Status == stellar1.PaymentStatus_PENDING {
				// add the tx id and the index of this payment to a waiting list
				waiting[bpResult.TxID] = i
			}
		}

		bpResult.StatusDescription = stellar1.PaymentStatusRevMap[bpResult.Status]
		resultList[i] = bpResult
	}

	// release the lock before waiting for the payments
	unlock()

	res.AllSubmittedTime = stellar1.ToTimeMs(time.Now())

	// wait for the payments
	waitingCount := len(waiting)
	mctx.Debug("waiting for %d payments to complete", waitingCount)

	timedOut := false
	var chatWaitGroup sync.WaitGroup
	for waitingCount > 0 && !timedOut {
		select {
		case <-time.After(5 * time.Second):
			if time.Since(startTime) > time.Duration(arg.TimeoutSecs)*time.Second {
				mctx.Debug("ran out of time waiting for tx status updates (%d remaining)", waitingCount)
				timedOut = true
			}
		case update := <-listenerCh:
			index, ok := waiting[update.TxID]
			if ok {
				mctx.Debug("received status update for %s: %s", update.TxID, update.Status)
				resultList[index].Status = update.Status
				resultList[index].StatusDescription = stellar1.PaymentStatusRevMap[update.Status]
				if update.Status != stellar1.PaymentStatus_PENDING {
					waitingCount--
					resultList[index].EndTime = stellar1.ToTimeMs(time.Now())
					delete(waiting, update.TxID)
					mctx.Debug("no longer waiting for %s status updates (%d remaining)", update.TxID, waitingCount)
				}
				if update.Status == stellar1.PaymentStatus_COMPLETED || update.Status == stellar1.PaymentStatus_CLAIMABLE {
					chatWaitGroup.Add(1)
					go func(m libkb.MetaContext, recipient string, txID stellar1.TransactionID) {
						if err := chatSendPaymentMessageTo(m, recipient, txID); err != nil {
							m.Debug("chatSendPaymentMessageTo %s (%s): error: %s", recipient, txID, err)
						} else {
							m.Debug("chatSendPaymentMessageTo %s (%s): success", recipient, txID)
						}

						chatWaitGroup.Done()
					}(mctx.WithCtx(context.Background()), resultList[index].Username, update.TxID)
				}
			}
		}
	}

	res.AllCompleteTime = stellar1.ToTimeMs(time.Now())
	mctx.Debug("done waiting for payments to complete")

	mctx.Debug("waiting for chat messages to finish sending")
	chatWaitGroup.Wait()
	mctx.Debug("done waiting for chat messages to finish sending")

	res.Payments = resultList
	res.EndTime = stellar1.ToTimeMs(time.Now())
	calculateStats(&res)

	return res, nil
}
490286
func (b *AllocUpdateBatcher) batcher() {
	var timerCh <-chan time.Time
	allocs := make(map[string]*structs.DesiredTransition)
	evals := make(map[string]*structs.Evaluation)
	future := NewBatchFuture()
	for {
		select {
		case <-b.ctx.Done():
			return
		case w := <-b.workCh:
			if timerCh == nil {
				timerCh = time.After(b.batch)
			}

			// Store the eval and alloc updates, and attach the future
			evals[w.e.DeploymentID] = w.e
			for id, upd := range w.allocs {
				allocs[id] = upd
			}

			w.f <- future
		case <-timerCh:
			// Capture the future and create a new one
			f := future
			future = NewBatchFuture()

			// Shouldn't be possible
			if f == nil {
				panic("no future")
			}

			// Create the request
			req := &structs.AllocUpdateDesiredTransitionRequest{
				Allocs: allocs,
				Evals:  make([]*structs.Evaluation, 0, len(evals)),
			}

			for _, e := range evals {
				req.Evals = append(req.Evals, e)
			}

			// Upsert the evals in a go routine
			go f.Set(b.raft.UpdateAllocDesiredTransition(req))

			// Reset the evals list and timer
			evals = make(map[string]*structs.Evaluation)
			allocs = make(map[string]*structs.DesiredTransition)
			timerCh = nil
		}
	}
}
878601
func (mvcc *MVCCLevelDB) RawBatchDelete(keys [][]byte) {
	mvcc.mu.Lock()
	defer mvcc.mu.Unlock()

	batch := &leveldb.Batch{}
	for _, key := range keys {
		batch.Delete(key)
	}
	terror.Log(mvcc.db.Write(batch, nil))
}
1351211
func (b *batch) Put(key, value []byte) error {
	b.b.Put(key, value)
	b.size += len(value)
	return nil
}
1333253
func (b *FailBatch) Put(k ds.Key, val []byte) error {
	if err := b.dstore.errfunc("batch-put"); err != nil {
		return err
	}

	return b.cb.Put(k, val)
}
258061
func (dds *Delayed) Batch() (ds.Batch, error) {
	return ds.NewBasicBatch(dds), nil
}
722639
func (wb *WriteBatch) BatchData() *BatchData {
	data := wb.wb.Data()
	if wb.data == nil {
		wb.data = new(BatchData)
	}

	wb.data.Load(data)
	return wb.data
}
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
}
100087
func newBatch(rootCollection *collection, options BatchOptions) (
	*batch, error) {
	return &batch{
		segment: &segment{
			kvs:            make([]uint64, 0, options.TotalOps*2),
			buf:            make([]byte, 0, options.TotalKeyValBytes),
			rootCollection: rootCollection,
		},
		childBatches: nil, // Created later on demand.
	}, nil
}
1074813
func (batch *Batch) Put(cmd string, args ...interface{}) error {
    if len(args) < 1 {
	return fmt.Errorf("Put: no key found in args")
    }

    if cmd == "MGET" || cmd == "MSET" || cmd == "MSETNX" {
	return fmt.Errorf("Put: %s not supported", cmd)
    }

    node, err := batch.cluster.getNodeByKey(args[0])
    if err != nil {
	return fmt.Errorf("Put: %v", err)
    }

    var i int
    for i = 0; i < len(batch.batches); i++ {
	if batch.batches[i].node == node {
	    batch.batches[i].cmds = append(batch.batches[i].cmds,
		nodeCommand{cmd: cmd, args: args})

	    batch.index = append(batch.index, i)
	    break
	}
    }

    if i == len(batch.batches) {
	batch.batches = append(batch.batches,
	    nodeBatch{
		node: node,
		cmds: []nodeCommand{{cmd: cmd, args: args}},
		done: make(chan int)})
	batch.index = append(batch.index, i)
    }

    return nil
}
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
}
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)
}
707878
func (mBatch *goLevelDBBatch) Set(key, value []byte) {
	mBatch.batch.Put(key, value)
}
699572
func (b *Batch) Add(doc interface{}) {
	b.Docs = append(b.Docs, doc)
}
1298547
func (clnt *Client) batchIndexExecute(policy *BatchPolicy, records []*BatchRead, cmd batchIndexer) error {
	batchNodes, err := newBatchIndexNodeList(clnt.cluster, policy, records)
	if err != nil {
		return err
	}

	var wg sync.WaitGroup

	// Use a goroutine per namespace per node
	errs := []error{}
	errm := new(sync.Mutex)

	wg.Add(len(batchNodes))
	for _, batchNode := range batchNodes {
		// copy to avoid race condition
		newCmd := cmd.cloneBatchIndexCommand(batchNode)
		go func(cmd command) {
			defer wg.Done()
			if err := cmd.Execute(); err != nil {
				errm.Lock()
				errs = append(errs, err)
				errm.Unlock()
			}
		}(newCmd)
	}

	wg.Wait()
	return mergeErrors(errs)
}