// Implements Batch.
541533
func (dbch debugBatch) Set(key, value []byte) {
	fmt.Printf("%v.batch.Set(%v, %v)\n", dbch.label,
		cmn.ColoredBytes(key, cmn.Yellow, cmn.Blue),
		cmn.ColoredBytes(value, cmn.Green, cmn.Blue))
	dbch.bch.Set(key, value)
}
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)
}
1217320
func (m SessionMock) Batch(kind BatchKind) Batch {
	return m.Called(kind).Get(0).(Batch)
}
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
}
84228
func (mvcc *MVCCLevelDB) RawBatchPut(keys, values [][]byte) {
	mvcc.mu.Lock()
	defer mvcc.mu.Unlock()

	batch := &leveldb.Batch{}
	for i, key := range keys {
		value := values[i]
		if value == nil {
			value = []byte{}
		}
		batch.Put(key, value)
	}
	terror.Log(mvcc.db.Write(batch, nil))
}
1114235
func (db *MemDB) NewBatch() Batch {
	db.mtx.Lock()
	defer db.mtx.Unlock()

	return &memBatch{db, nil}
}
1062621
func (n node) WithBatch(enabled bool) Node {
	n.batchMode = enabled
	return &n
}
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
}
1324393
func (rl *MemoryReplayLog) PutBatch(batch *Batch) (*ReplaySet, error) {
	if rl.entries == nil || rl.batches == nil {
		return nil, errReplayLogNotStarted
	}

	// Return the result when the batch was first processed to provide
	// idempotence.
	replays, exists := rl.batches[string(batch.ID)]

	if !exists {
		replays = NewReplaySet()
		err := batch.ForEach(func(seqNum uint16, hashPrefix *HashPrefix, cltv uint32) error {
			err := rl.Put(hashPrefix, cltv)
			if err == ErrReplayedPacket {
				replays.Add(seqNum)
				return nil
			}

			// An error would be bad because we have already updated the entries
			// map, but no errors other than ErrReplayedPacket should occur.
			return err
		})
		if err != nil {
			return nil, err
		}

		replays.Merge(batch.ReplaySet)
		rl.batches[string(batch.ID)] = replays
	}

	batch.ReplaySet = replays
	batch.IsCommitted = true

	return replays, nil
}
105104
func (p *AgentClient) EmitBatch(batch *jaeger.Batch) (err error) {
	if err = p.sendEmitBatch(batch); err != nil {
		return
	}
	return
}
547225
func (cluster *Cluster) RunBatch(bat *Batch) ([]interface{}, error) {
    for i := range bat.batches {
	go doBatch(&bat.batches[i])
    }

    for i := range bat.batches {
	<-bat.batches[i].done
    }

    var replies []interface{}
    for _, i := range bat.index {
	if bat.batches[i].err != nil {
	    return nil, bat.batches[i].err
	}

	replies = append(replies, bat.batches[i].cmds[0].reply)
	bat.batches[i].cmds = bat.batches[i].cmds[1:]
    }

    return replies, nil
}
58395
func (dbch debugBatch) WriteSync() {
	fmt.Printf("%v.batch.WriteSync()\n", dbch.label)
	dbch.bch.WriteSync()
}
460599
func (mBatch *cLevelDBBatch) Set(key, value []byte) {
	mBatch.batch.Put(key, value)
}
1044934
func (mBatch *cLevelDBBatch) Write() {
	err := mBatch.db.db.Write(mBatch.db.wo, mBatch.batch)
	if err != nil {
		panic(err)
	}
}
1351211
func (b *batch) Put(key, value []byte) error {
	b.b.Put(key, value)
	b.size += len(value)
	return nil
}
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
}
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
}