// GetSchedulableMulticastQueueItems returns a slice of multicast-queue items
// for scheduling.
// The returned queue-items will be locked for update so that this query can
// be executed in parallel.
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
}
1272326
func (pq *PriorityQueue) Get(number int) ([]Item, error) {
	if number < 1 {
		return nil, nil
	}

	pq.lock.Lock()

	if pq.disposed {
		pq.lock.Unlock()
		return nil, ErrDisposed
	}

	var items []Item

	// Remove references to popped items.
	deleteItems := func(items []Item) {
		for _, item := range items {
			delete(pq.itemMap, item)
		}
	}

	if len(pq.items) == 0 {
		sema := newSema()
		pq.waiters.put(sema)
		pq.lock.Unlock()

		<-sema.ready

		if pq.Disposed() {
			return nil, ErrDisposed
		}

		items = pq.items.get(number)
		if !pq.allowDuplicates {
			deleteItems(items)
		}
		sema.response.Done()
		return items, nil
	}

	items = pq.items.get(number)
	deleteItems(items)
	pq.lock.Unlock()
	return items, nil
}
1289478
func (tc *TransCache) GetGroupItems(chID, grpID string) (itms []interface{}) {
	tc.cacheMux.RLock()
	itms = tc.cacheInstance(chID).GetGroupItems(grpID)
	tc.cacheMux.RUnlock()
	return
}
1294394
func (r Billing_Item_Virtual_DedicatedHost) GetActiveCancellationItem() (resp datatypes.Billing_Item_Cancellation_Request_Item, err error) {
	err = r.Session.DoRequest("SoftLayer_Billing_Item_Virtual_DedicatedHost", "getActiveCancellationItem", nil, &r.Options, &resp)
	return
}
223154
func (c *ElastiCache) WaitUntilCacheClusterAvailable(input *DescribeCacheClustersInput) error {
	return c.WaitUntilCacheClusterAvailableWithContext(aws.BackgroundContext(), input)
}
854286
func (c *Client) GetAllScheduledMaintenances() (ScheduledMaintenancesResponse, error) {
	s := ScheduledMaintenancesResponse{}
	err := c.request("api/v2/scheduled-maintenances.json", &s)
	return s, err
}
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())
}
458473
func (tc *TransCache) GetGroupItemIDs(chID, grpID string) (itmIDs []string) {
	tc.cacheMux.RLock()
	itmIDs = tc.cacheInstance(chID).GetGroupItemIDs(grpID)
	tc.cacheMux.RUnlock()
	return
}
596677
func (j *Jenkins) GetQueueItem(id int64) (*Task, error) {
	t := &Task{Raw: new(taskResponse), Jenkins: j, Base: j.getQueueItemURL(id)}
	_, err := t.Poll()
	if err != nil {
		return nil, err
	}
	return t, nil
}
458021
func (c *Cache) GetItemIDs(prfx string) (itmIDs []string) {
	c.RLock()
	for itmID := range c.cache {
		if strings.HasPrefix(itmID, prfx) {
			itmIDs = append(itmIDs, itmID)
		}
	}
	c.RUnlock()
	return
}
414494
func listScheduledJobsAction(client *gocd.Client, c *cli.Context) (r interface{}, resp *gocd.APIResponse, err error) {
	return client.Jobs.ListScheduled(context.Background())
}
675393
func HandleScheduleNextQueueItem(ds storage.DeviceSession, mode storage.DeviceMode) error {
	ctx := dataContext{
		DeviceMode:    mode,
		DeviceSession: ds,
	}

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

	return nil
}
558133
func (n *NetworkServerAPI) EnqueueMulticastQueueItem(ctx context.Context, req *ns.EnqueueMulticastQueueItemRequest) (*empty.Empty, error) {
	if req.MulticastQueueItem == nil {
		return nil, grpc.Errorf(codes.InvalidArgument, "multicast_queue_item must not be nil")
	}

	var mgID uuid.UUID
	copy(mgID[:], req.MulticastQueueItem.MulticastGroupId)

	qi := storage.MulticastQueueItem{
		MulticastGroupID: mgID,
		FCnt:             req.MulticastQueueItem.FCnt,
		FPort:            uint8(req.MulticastQueueItem.FPort),
		FRMPayload:       req.MulticastQueueItem.FrmPayload,
	}

	err := storage.Transaction(func(tx sqlx.Ext) error {
		return multicast.EnqueueQueueItem(storage.RedisPool(), tx, qi)
	})
	if err != nil {
		return nil, errToRPCError(err)
	}

	return &empty.Empty{}, nil
}
713671
func NewScheduler() (*Scheduler, error) {
	defaultClusterOperations := map[string]bool{
		"HorizontalReshardingTask": true,
		"VerticalSplitTask":        true,
	}

	s := &Scheduler{
		registeredClusterOperations:    defaultClusterOperations,
		idGenerator:                    IDGenerator{},
		toBeScheduledClusterOperations: make(chan ClusterOperationInstance, 10),
		state:                          stateNotRunning,
		taskCreator:                    defaultTaskCreator,
		pendingOpsWg:                   &sync.WaitGroup{},
		activeClusterOperations:        make(map[string]ClusterOperationInstance),
		finishedClusterOperations:      make(map[string]ClusterOperationInstance),
	}

	return s, nil
}
565450
func (s *Scheduler) EnqueueClusterOperation(ctx context.Context, req *automationpb.EnqueueClusterOperationRequest) (*automationpb.EnqueueClusterOperationResponse, error) {
	s.mu.Lock()
	defer s.mu.Unlock()

	if s.state != stateRunning {
		return nil, fmt.Errorf("scheduler is not running. State: %v", s.state)
	}

	if !s.registeredClusterOperations[req.Name] {
		return nil, fmt.Errorf("no ClusterOperation with name: %v is registered", req.Name)
	}

	err := s.validateTaskSpecification(req.Name, req.Parameters)
	if err != nil {
		return nil, err
	}

	clusterOpID := s.idGenerator.GetNextID()
	taskIDGenerator := IDGenerator{}
	initialTask := NewTaskContainerWithSingleTask(req.Name, req.Parameters)
	clusterOp := NewClusterOperationInstance(clusterOpID, initialTask, &taskIDGenerator)

	s.muOpList.Lock()
	s.activeClusterOperations[clusterOpID] = clusterOp.Clone()
	s.muOpList.Unlock()
	s.toBeScheduledClusterOperations <- clusterOp

	return &automationpb.EnqueueClusterOperationResponse{
		Id: clusterOp.Id,
	}, nil
}
40551
func (pq *PriorityQueue) getNextItem() (*PriorityItem, error) {
	// If the current priority level is empty.
	if pq.levels[pq.curLevel].length() == 0 {
		// Set starting value for curLevel.
		pq.resetCurrentLevel()

		// Try to get the next priority level.
		for i := 0; i <= 255; i++ {
			if (pq.cmpAsc(uint8(i)) || pq.cmpDesc(uint8(i))) && pq.levels[uint8(i)].length() > 0 {
				pq.curLevel = uint8(i)
			}
		}

		// If still empty, return queue empty error.
		if pq.levels[pq.curLevel].length() == 0 {
			return nil, ErrEmpty
		}
	}

	// Try to get the next item in the current priority level.
	return pq.getItemByPriorityID(pq.curLevel, pq.levels[pq.curLevel].head+1)
}
906269
func (s *StanServer) sendAvailableMessagesToQueue(c *channel, qs *queueState) {
	if c == nil || qs == nil {
		return
	}

	qs.Lock()
	// Short circuit if no active members
	if len(qs.subs) == 0 {
		qs.Unlock()
		return
	}
	// If redelivery at startup in progress, don't attempt to deliver new messages
	if qs.newOnHold {
		qs.Unlock()
		return
	}
	for nextSeq := qs.lastSent + 1; qs.stalledSubCount < len(qs.subs); nextSeq++ {
		nextMsg := s.getNextMsg(c, &nextSeq, &qs.lastSent)
		if nextMsg == nil {
			break
		}
		if _, sent, sendMore := s.sendMsgToQueueGroup(qs, nextMsg, honorMaxInFlight); !sent || !sendMore {
			break
		}
	}
	qs.Unlock()
}