// NewScheduler returns a new scheduler with the given desired state and the given now UTC timestamp.
346254
func NewScheduler(desiredState DesiredState, executor Executor, lw LogWriter, now int64, opts ...TickSchedulerOption) *TickScheduler {
	o := &TickScheduler{
		desiredState:   desiredState,
		executor:       executor,
		logWriter:      lw,
		now:            now,
		taskSchedulers: make(map[platform.ID]*taskScheduler),
		logger:         zap.NewNop(),
		wg:             &sync.WaitGroup{},
		metrics:        newSchedulerMetrics(),
	}

	for _, opt := range opts {
		opt(o)
	}

	return o
}
789584
func (s *CreateSnapshotScheduleInput) SetScheduleIdentifier(v string) *CreateSnapshotScheduleInput {
	s.ScheduleIdentifier = &v
	return s
}
1353026
func NewStateResourceManager(sd core.StateDriver) (*StateResourceManager, error) {
	if gStateResourceManager != nil {
		return nil, core.Errorf("state-based resource manager instance already exists.")
	}

	gStateResourceManager = &StateResourceManager{stateDriver: sd}
	err := gStateResourceManager.Init()
	if err != nil {
		return nil, err
	}

	return gStateResourceManager, nil
}
828439
func (m *Memberlist) schedule() {
	m.tickerLock.Lock()
	defer m.tickerLock.Unlock()

	// If we already have tickers, then don't do anything, since we're
	// scheduled
	if len(m.tickers) > 0 {
		return
	}

	// Create the stop tick channel, a blocking channel. We close this
	// when we should stop the tickers.
	stopCh := make(chan struct{})

	// Create a new probeTicker
	if m.config.ProbeInterval > 0 {
		t := time.NewTicker(m.config.ProbeInterval)
		go m.triggerFunc(m.config.ProbeInterval, t.C, stopCh, m.probe)
		m.tickers = append(m.tickers, t)
	}

	// Create a push pull ticker if needed
	if m.config.PushPullInterval > 0 {
		go m.pushPullTrigger(stopCh)
	}

	// Create a gossip ticker if needed
	if m.config.GossipInterval > 0 && m.config.GossipNodes > 0 {
		t := time.NewTicker(m.config.GossipInterval)
		go m.triggerFunc(m.config.GossipInterval, t.C, stopCh, m.gossip)
		m.tickers = append(m.tickers, t)
	}

	// If we made any tickers, then record the stopTick channel for
	// later.
	if len(m.tickers) > 0 {
		m.stopTick = stopCh
	}
}
395584
func (a *Allocation) ShouldReschedule(reschedulePolicy *ReschedulePolicy, failTime time.Time) bool {
	// First check the desired state
	switch a.DesiredStatus {
	case AllocDesiredStatusStop, AllocDesiredStatusEvict:
		return false
	default:
	}
	switch a.ClientStatus {
	case AllocClientStatusFailed:
		return a.RescheduleEligible(reschedulePolicy, failTime)
	default:
		return false
	}
}
1056394
func newSender(client metricsmanager.MetricsManagerClient, notify chan string) worker.Worker {
	f := func(stopCh <-chan struct{}) error {
		err := client.SendMetrics()
		if err != nil {
			senderLogger.Warningf("failed to send metrics %v - will retry later", err)
			return nil
		}
		select {
		case notify <- "senderCalled":
		default:
		}
		return nil
	}
	return jworker.NewPeriodicWorker(f, senderPeriod, jworker.NewTimer)
}
843135
func (s *UpdateMaintenanceWindowOutput) SetSchedule(v string) *UpdateMaintenanceWindowOutput {
	s.Schedule = &v
	return s
}
83128
func (w *WorkerConfig) scheduler() {
	pollSets := []string{w.nsKey("retry"), w.nsKey("schedule")}

	for _ = range time.Tick(w.PollInterval) {
		w.RLock() // don't let quitHandler() stop us in the middle of a run
		conn := w.RedisPool.Get()
		now := fmt.Sprintf("%f", timeFloat(time.Now()))
		for _, set := range pollSets {
			conn.Send("MULTI")
			conn.Send("ZRANGEBYSCORE", set, "-inf", now)
			conn.Send("ZREMRANGEBYSCORE", set, "-inf", now)
			res, err := redis.Values(conn.Do("EXEC"))
			if err != nil {
				w.handleError(err)
				continue
			}

			for _, msg := range res[0].([]interface{}) {
				parsedMsg := &struct {
					Queue string `json:"queue"`
				}{}
				msgBytes := msg.([]byte)
				err := json.Unmarshal(msgBytes, parsedMsg)
				if err != nil {
					w.handleError(err)
					continue
				}
				if _, err = conn.Do("RPUSH", w.nsKey("queue:"+parsedMsg.Queue), msgBytes); err != nil {
					w.handleError(err)
				}
			}
		}
		conn.Close()
		w.RUnlock()
	}
}
118335
func Scheduler(c context.Context, cfg *pb.ProjectConfig) error {
	// Sort jobs by ID.
	sort.Slice(cfg.Job, func(i, j int) bool {
		return cfg.Job[i].Id < cfg.Job[j].Id
	})
	sort.Slice(cfg.Trigger, func(i, j int) bool {
		return cfg.Trigger[i].Id < cfg.Trigger[j].Id
	})

	// ACL set name => list of ACLs.
	aclSets := make(map[string][]*pb.Acl)
	for _, s := range cfg.AclSets {
		aclSets[s.Name] = s.Acls
	}

	// Expand ACL sets.
	for _, x := range cfg.Job {
		x.Acls = normalizeACLs(x.Acls, x.AclSets, aclSets)
		x.AclSets = nil
	}
	for _, x := range cfg.Trigger {
		x.Acls = normalizeACLs(x.Acls, x.AclSets, aclSets)
		x.AclSets = nil
	}
	cfg.AclSets = nil

	// Normalize triggers.
	for _, x := range cfg.Trigger {
		sort.Strings(x.Triggers)
		if gt := x.Gitiles; gt != nil {
			for i, r := range gt.Refs {
				if !strings.HasPrefix(r, "regexp:") {
					gt.Refs[i] = "regexp:" + regexp.QuoteMeta(r)
				}
			}
			sort.Strings(gt.Refs)
		}
	}

	return nil
}
1227553
func (s *SnapshotSchedule) SetScheduleDescription(v string) *SnapshotSchedule {
	s.ScheduleDescription = &v
	return s
}
1322245
func (s *ModifyClusterSnapshotScheduleInput) SetScheduleIdentifier(v string) *ModifyClusterSnapshotScheduleInput {
	s.ScheduleIdentifier = &v
	return s
}
349813
func (s *PolicyDetails) SetSchedules(v []*Schedule) *PolicyDetails {
	s.Schedules = v
	return s
}
302795
func (m *Matcher) addScheduler(idx uint) {
	if _, ok := m.schedulers[idx]; ok {
		return
	}
	m.schedulers[idx] = newScheduler(idx)
}
66722
func NewScheduler(
	numScrapers int,
	l *zap.Logger,
	targets platform.ScraperTargetStoreService,
	p nats.Publisher,
	s nats.Subscriber,
	interval time.Duration,
	timeout time.Duration,
) (*Scheduler, error) {
	if interval == 0 {
		interval = 60 * time.Second
	}
	if timeout == 0 {
		timeout = 30 * time.Second
	}
	scheduler := &Scheduler{
		Targets:   targets,
		Interval:  interval,
		Timeout:   timeout,
		Publisher: p,
		Logger:    l,
		gather:    make(chan struct{}, 100),
	}

	for i := 0; i < numScrapers; i++ {
		err := s.Subscribe(promTargetSubject, "", &handler{
			Scraper:   new(prometheusScraper),
			Publisher: p,
			Logger:    l,
		})
		if err != nil {
			return nil, err
		}
	}

	return scheduler, nil
}
1042174
func (s *WriteTreatmentResource) SetSchedule(v *Schedule) *WriteTreatmentResource {
	s.Schedule = v
	return s
}
819347
func newStateSync(d *Downloader, root common.Hash) *stateSync {
	return &stateSync{
		d:       d,
		sched:   state.NewStateSync(root, d.stateDB),
		keccak:  sha3.NewLegacyKeccak256(),
		tasks:   make(map[common.Hash]*stateTask),
		deliver: make(chan *stateReq),
		cancel:  make(chan struct{}),
		done:    make(chan struct{}),
	}
}
1056600
func (s *BusinessReportSchedule) SetScheduleArn(v string) *BusinessReportSchedule {
	s.ScheduleArn = &v
	return s
}