// ScheduleMulticastQueueBatch schedules a donwlink multicast batch (Class-B & -C).
970453
func ScheduleMulticastQueueBatch(size int) error {
	return storage.Transaction(func(tx sqlx.Ext) error {
		// this locks the selected queue-items so that this query can be
		// executed by other instances in parallel.
		multicastQueueItems, err := storage.GetSchedulableMulticastQueueItems(tx, size)
		if err != nil {
			return errors.Wrap(err, "get multicast queue-items error")
		}

		for _, qi := range multicastQueueItems {
			err := multicast.HandleScheduleQueueItem(tx, qi)
			if err != nil {
				log.WithFields(log.Fields{
					"multicast_group_id": qi.MulticastGroupID,
					"id":                 qi.ID,
				}).WithError(err).Error("schedule multicast queue-item error")
			}
		}

		return nil
	})
}
1062621
func (n node) WithBatch(enabled bool) Node {
	n.batchMode = enabled
	return &n
}
770220
func GetSchedulableMulticastQueueItems(db sqlx.Ext, count int) ([]MulticastQueueItem, error) {
	var items []MulticastQueueItem
	err := sqlx.Select(db, &items, `
		select
			*
		from
			multicast_queue
		where
			schedule_at <= $2
		order by
			id
		limit $1
		for update skip locked
	`, count, time.Now())
	if err != nil {
		return nil, handlePSQLError(err, "select error")
	}

	return items, nil
}
3521
func (c *Client) ScheduleAppGarbageCollection(appID string) error {
	return c.Post(fmt.Sprintf("/apps/%s/gc", appID), nil, nil)
}
814515
func (d *CacheDumper) dumpSchedulingQueue() {
	pendingPods := d.podQueue.PendingPods()
	var podData strings.Builder
	for _, p := range pendingPods {
		podData.WriteString(printPod(p))
	}
	klog.Infof("Dump of scheduling queue:\n%s", podData.String())
}
16378
func (c *Client) MultipleBatch(operations []BatchOperationIndexed, opts ...interface{}) (res MultipleBatchRes, err error) {
	body := map[string][]BatchOperationIndexed{"requests": operations}
	path := c.path("/indexes/*/batch")
	err = c.transport.Request(&res, http.MethodPost, path, body, call.Write, opts...)
	res.wait = c.waitTask
	return
}
907445
func (dq *deleteQueue) deleteBatch(msgs []*sqs.DeleteMessageBatchRequestEntry) ([]*sqs.DeleteMessageBatchRequestEntry, error) {
	req := &sqs.DeleteMessageBatchInput{
		QueueUrl: dq.svc.URL,
		Entries:  msgs,
	}

	resp, err := dq.svc.Svc.DeleteMessageBatch(req)
	if err != nil {
		dq.svc.Logger("Error deleting messages: %s", err)
		return nil, err
	}

	var failed []*sqs.DeleteMessageBatchRequestEntry
	for _, f := range resp.Failed {
		for _, m := range msgs {
			if *m.Id == *f.Id {
				failed = append(failed, m)
				break
			}
		}
	}
	return failed, nil
}
1249304
func (db *CLevelDB) NewBatch() Batch {
	batch := levigo.NewWriteBatch()
	return &cLevelDBBatch{db, batch}
}
952768
func NewAllocUpdateBatcher(batchDuration time.Duration, raft DeploymentRaftEndpoints, ctx context.Context) *AllocUpdateBatcher {
	b := &AllocUpdateBatcher{
		batch:  batchDuration,
		raft:   raft,
		ctx:    ctx,
		workCh: make(chan *updateWrapper, 10),
	}

	go b.batcher()
	return b
}
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
}
1216508
func DownlinkQueue(prefix string) storage.MigrateFunction {
	return func(client *redis.Client, key string, obj map[string]string) (string, map[string]string, error) {
		var err error
		nextDownlink, ok := obj["next_downlink"]
		if ok {
			delete(obj, "next_downlink")
			scheduleKey := prefix + ":downlink" + strings.TrimPrefix(key, prefix+":device")
			err = client.LPush(scheduleKey, nextDownlink).Err()
		}
		return "2.4.2", obj, err
	}
}
767965
func (cluster *Cluster) NewBatch() *Batch {
    return &Batch{
	cluster: cluster,
	batches: make([]nodeBatch, 0),
	index: make([]int, 0),
    }
}
1166449
func HandleScheduleNextQueueItem(db sqlx.Ext, mg storage.MulticastGroup) error {
	ctx := multicastContext{
		DB:             db,
		MulticastGroup: mg,
	}

	for _, t := range multicastTasks {
		if err := t(&ctx); err != nil {
			if err == errAbort {
				return nil
			}
			return err
		}
	}

	return nil
}
618355
func NewTaskQueue(syncFn func(string) error) *taskQueue {
	return &taskQueue{
		queue:      workqueue.NewRateLimitingQueue(workqueue.DefaultControllerRateLimiter()),
		sync:       syncFn,
		workerDone: make(chan struct{}),
	}
}
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
}
997474
func (t *Template) GetAllAWSBatchJobQueueResources() map[string]*resources.AWSBatchJobQueue {
	results := map[string]*resources.AWSBatchJobQueue{}
	for name, untyped := range t.Resources {
		switch resource := untyped.(type) {
		case *resources.AWSBatchJobQueue:
			results[name] = resource
		}
	}
	return results
}
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()
	}
}