// ScheduleServices merges and reconciles a slice of services with the
// ServiceStateChangeBatches in the ServiceStateManager's queue
497112
func (s *BatchServiceStateManager) ScheduleServices(svcs []*service.Service, tenantID string, desiredState service.DesiredState, emergency bool) error {
	logger := plog.WithFields(logrus.Fields{
		"servicecount": len(svcs),
		"tenantid":     tenantID,
		"desiredstate": desiredState.String(),
		"emergency":    emergency,
	})

	var err error
	logger.Debug("Scheduling services")
	s.lock.Lock()
	var queues map[service.DesiredState]*ServiceStateQueue
	queues, ok := s.TenantQueues[tenantID]

	if !ok {
		s.lock.Unlock()
		return ErrBadTenantID
	}

	// Get a lock on all queues
	for _, q := range queues {
		q.lock.Lock()
		defer q.lock.Unlock()
	}
	s.lock.Unlock()
	// Build the cancellable services from the list
	cancellableServices := make(map[string]*CancellableService)
	for _, svc := range svcs {
		cancellableServices[svc.ID] = NewCancellableService(svc)
	}

	// Merge with oldBatch batchQueue
	// 1. If this is emergency, merge with other emergencies and move to front of the queue
	// 2. If any service in this batch is currently in the "pending" batch:
	//    A. If the desired states are the same, leave it pending and remove it from this batch
	//    B. If the desired states are different, cancel waits on the pending request and leave it in this batch
	// 3. If this and the last N batches at the end of the queue all have the same desired state, merge and re-group them
	// 4. If any service in this batch also appears in an earlier batch:
	//    A. If the desired state is the same, leave it in the earlier batch and remove it here
	//    B. If the desired state is different, delete it from the earlier batch and leave it in the new one
	newBatch := ServiceStateChangeBatch{
		Services:     cancellableServices,
		DesiredState: desiredState,
		Emergency:    emergency,
	}
	var expeditedBatch ServiceStateChangeBatch
	expeditedServices := make(map[string]*CancellableService)
	var cancelledIDs []string
	for _, queue := range queues {
		// reconcile the new batch against all batches in q
		var cancelled []string
		newBatch, expeditedBatch, cancelled = queue.reconcileWithBatchQueue(newBatch)
		cancelledIDs = append(cancelledIDs, cancelled...)
		for id, svc := range expeditedBatch.Services {
			expeditedServices[id] = svc
		}

		if len(newBatch.Services) == 0 {
			// Nothing left to reconcile
			break
		}

		// reconcile with the pending batch
		newBatch = queue.reconcileWithPendingBatch(newBatch)
	}

	// Now check for services that are already in the desired state
	var cancelled []string
	newBatch, cancelled = s.reconcileWithCurrentState(newBatch)
	cancelledIDs = append(cancelledIDs, cancelled...)

	if len(cancelledIDs) > 0 {
		// Spawn a goroutine to update the current state of all cancelled services
		defer func() {
			logger.WithField("services", cancelledIDs).Debug("Cancelling and re-syncing services")
			go s.SyncCurrentStates(cancelledIDs)
		}()
	}

	expeditedBatch.Services = expeditedServices

	if len(expeditedBatch.Services) > 0 {
		// process the expedited batch now
		go func() {
			s.updateBatch(&expeditedBatch)
			failed := s.processBatch(tenantID, expeditedBatch)
			// Cancel these services to notify waiters that they have been scheduled
			for _, svc := range expeditedBatch.Services {
				svc.Cancel()
			}
			if len(failed) > 0 {
				s.SyncCurrentStates(failed)
			}
		}()
	}

	if len(newBatch.Services) == 0 {
		return nil
	}

	queueDesiredState := newBatch.DesiredState
	switch newBatch.DesiredState {
	case service.SVCRestart:
		queueDesiredState = service.SVCRun
	case service.SVCPause:
		queueDesiredState = service.SVCStop
	}

	// Merge this into the queue
	queue, ok := queues[queueDesiredState]
	if !ok {
		return ErrMissingQueue
	}

	if newBatch.Emergency {
		err = queue.mergeEmergencyBatch(newBatch)
	} else {
		err = queue.mergeBatch(newBatch)
	}

	// Set pending desired state
	currentState := service.DesiredToCurrentPendingState(newBatch.DesiredState, newBatch.Emergency)
	var sids []string
	for sid := range newBatch.Services {
		sids = append(sids, sid)
	}
	s.currentStateLock.Lock()
	s.updateServiceCurrentState(currentState, sids...)
	s.currentStateLock.Unlock()

	// Signal update or exit
	select {
	case queue.Changed <- true:
	default:
	}

	return err
}
556953
func (s *ModifySnapshotScheduleOutput) SetScheduleDefinitions(v []*string) *ModifySnapshotScheduleOutput {
	s.ScheduleDefinitions = v
	return s
}
1140606
func (s *Supervisor) Services() map[string]Service {
	svclist := make(map[string]Service)
	s.mu.Lock()
	for k, v := range s.services {
		svclist[k] = v.svc
	}
	s.mu.Unlock()
	return svclist
}
547787
func ListServices(context Context, wait time.Duration, log Log) (*keybase1.ServicesStatus, error) {
	services, err := launchd.ListServices([]string{"keybase.service", "homebrew.mxcl.keybase"})
	if err != nil {
		return nil, err
	}
	kbfs, err := launchd.ListServices([]string{"keybase.kbfs.", "homebrew.mxcl.kbfs"})
	if err != nil {
		return nil, err
	}
	updater, err := launchd.ListServices([]string{"keybase.updater."})
	if err != nil {
		return nil, err
	}

	return &keybase1.ServicesStatus{
		Service: serviceStatusesFromLaunchd(context, services, wait, log),
		Kbfs:    serviceStatusesFromLaunchd(context, kbfs, wait, log),
		Updater: serviceStatusesFromLaunchd(context, updater, wait, log),
	}, nil
}
575626
func newScheduler(idx uint) *scheduler {
	return &scheduler{
		bit:       idx,
		responses: make(map[uint64]*response),
	}
}
1003206
func (s *EventTypeFilter) SetServices(v []*string) *EventTypeFilter {
	s.Services = v
	return s
}
667793
func (s *HTTPServer) OperatorSchedulerConfiguration(resp http.ResponseWriter, req *http.Request) (interface{}, error) {
	// Switch on the method
	switch req.Method {
	case "GET":
		return s.schedulerGetConfig(resp, req)

	case "PUT", "POST":
		return s.schedulerUpdateConfig(resp, req)

	default:
		return nil, CodedError(405, ErrInvalidMethod)
	}
}
884253
func (s *Scheduler) Services(appID string) (map[string]string, error) {
	stackName, err := s.stackName(appID)
	if err != nil {
		return nil, err
	}

	stack, err := s.stack(aws.String(stackName))
	if err != nil {
		return nil, fmt.Errorf("error describing stack: %v", err)
	}

	o := output(stack, servicesOutput)
	if o == nil {
		// Nothing to do but wait until the outputs are set.
		if *stack.StackStatus == cloudformation.StackStatusCreateInProgress {
			return nil, nil
		}

		return nil, fmt.Errorf("stack didn't provide a \"%s\" output key", servicesOutput)
	}

	return extractProcessData(*o.OutputValue), nil
}
1267163
func (s *DescribeScheduledActionsOutput) SetScheduledActions(v []*ScheduledAction) *DescribeScheduledActionsOutput {
	s.ScheduledActions = v
	return s
}
349813
func (s *PolicyDetails) SetSchedules(v []*Schedule) *PolicyDetails {
	s.Schedules = v
	return s
}
1336936
func SetServices(svcs []*Service) error {
	if defaultDevice == nil {
		return ErrDefaultDevice
	}
	return defaultDevice.SetServices(svcs)
}
717793
func MergeBatches(batches []ServiceStateChangeBatch) ([]ServiceStateChangeBatch, error) {
	if len(batches) < 1 {
		return batches, nil
	}

	var fullServiceList []*CancellableService

	// Make sure all of the batches we're merging have the same desiredState
	// and emergency status
	desiredState := batches[0].DesiredState
	emergency := batches[0].Emergency
	for _, b := range batches {
		if b.DesiredState != desiredState || b.Emergency != emergency {
			return nil, ErrMismatchedDesiredStates
		}
		for _, svc := range b.Services {
			fullServiceList = append(fullServiceList, svc)
		}
	}

	// Sort the full list based on desired state
	if desiredState == service.SVCRun {
		// Sort the services by start level
		sort.Sort(ByStartLevel{
			CancellableServices: fullServiceList,
		})
	} else if desiredState == service.SVCStop && emergency {
		sort.Sort(ByEmergencyShutdown{
			CancellableServices: fullServiceList,
		})
	} else if desiredState == service.SVCStop {
		sort.Sort(ByReverseStartLevel{
			CancellableServices: fullServiceList,
		})
	} else if desiredState == service.SVCRestart {
		sort.Sort(ByStartLevel{
			CancellableServices: fullServiceList,
		})
	} else if desiredState == service.SVCPause {
		sort.Sort(ByReverseStartLevel{
			CancellableServices: fullServiceList,
		})
	}

	// regroup the services by level
	previousEmergencyLevel := fullServiceList[0].EmergencyShutdownLevel
	previousStartLevel := fullServiceList[0].StartLevel

	newBatches := []ServiceStateChangeBatch{}
	newSvcs := make(map[string]*CancellableService)

	for _, svc := range fullServiceList {
		currentEmergencyLevel := svc.EmergencyShutdownLevel
		currentStartLevel := svc.StartLevel
		var sameBatch bool

		if emergency && desiredState == service.SVCStop {
			// this service should be in the same batch if it has the same emergency level
			sameBatch = currentEmergencyLevel == previousEmergencyLevel
			if sameBatch && currentEmergencyLevel == 0 {
				// For emergency shutdown level 0, we group by reverse start level
				sameBatch = currentStartLevel == previousStartLevel
			}
		} else {
			// this service should be in the same batch if it has the same start level
			sameBatch = currentStartLevel == previousStartLevel
		}

		if sameBatch {
			// add it to newSvcs
			newSvcs[svc.ID] = svc
		} else {
			// append our batch from the previous level, start a new batch with svc
			newBatches = append(newBatches, ServiceStateChangeBatch{
				Services:     newSvcs,
				DesiredState: desiredState,
				Emergency:    emergency,
			})
			newSvcs = make(map[string]*CancellableService)
			newSvcs[svc.ID] = svc
		}
		previousEmergencyLevel = currentEmergencyLevel
		previousStartLevel = currentStartLevel
	}

	// Add the last batch
	newBatches = append(newBatches, ServiceStateChangeBatch{
		Services:     newSvcs,
		DesiredState: desiredState,
		Emergency:    emergency,
	})

	return newBatches, nil
}
779467
func (s *ScheduleAction) SetScheduleActionStartSettings(v *ScheduleActionStartSettings) *ScheduleAction {
	s.ScheduleActionStartSettings = v
	return s
}
98219
func (state *ServicesState) BroadcastServices(fn func() []service.Service, looper director.Looper) {
	lastTime := time.Unix(0, 0)

	looper.Loop(func() error {
		defer metrics.MeasureSince([]string{"services_state", "BroadcastServices"}, time.Now())
		var services []service.Service
		haveNewServices := false

		servicesList := fn()

		state.RLock()
		defer state.RUnlock()

		for _, svc := range servicesList {
			isNew := state.IsNewService(&svc)

			// We'll broadcast it now if it's new or we've hit refresh window
			if isNew {
				log.Debug("Found service changes in BroadcastServices()")
				haveNewServices = true
				services = append(services, svc)
				// Check that refresh window... is it time?
			} else if time.Now().UTC().Add(0 - ALIVE_BROADCAST_INTERVAL).After(lastTime) {
				services = append(services, svc)
			}
		}

		if len(services) > 0 {
			log.Debug("Starting to broadcast")
			// Figure out how many times to announce the service. New services get more announcements.
			runCount := 1
			if haveNewServices {
				runCount = ALIVE_COUNT
			}

			lastTime = time.Now().UTC()
			state.SendServices(
				services,
				director.NewTimedLooper(runCount, state.tombstoneRetransmit, nil),
			)
			log.Debug("Completing broadcast")
		} else {
			// We expect there to always be _something_ in the channel
			// once we've run.
			state.Broadcasts <- nil
		}

		return nil
	})
}
1156387
func (state *ServicesState) SendServices(services []service.Service, looper director.Looper) {
	// Announce these every second for awhile
	go func() {
		defer metrics.MeasureSince([]string{"services_state", "SendServices"}, time.Now())

		additionalTime := 0 * time.Second
		looper.Loop(func() error {
			var prepared [][]byte

			for _, svc := range services {
				svc.Updated = svc.Updated.Add(additionalTime)
				encoded, err := svc.Encode()
				if err != nil {
					log.Errorf("ERROR encoding container: (%s)", err.Error())
				}
				prepared = append(prepared, encoded)
			}

			// We add time to make sure that these get retransmitted by peers.
			// Otherwise they aren't "new" messages and don't get retransmitted.
			additionalTime = additionalTime + 50*time.Nanosecond
			state.Broadcasts <- prepared // Put it on the wire
			return nil
		})
	}()
}
295279
func (s *ScheduleAction) SetScheduleActionSettings(v *ScheduleActionSettings) *ScheduleAction {
	s.ScheduleActionSettings = v
	return s
}
592707
func (s *scheduler) scheduleRequests(reqs chan uint64, dist chan *request, pend chan uint64, quit chan struct{}, wg *sync.WaitGroup) {
	// Clean up the goroutine and pipeline when done
	defer wg.Done()
	defer close(pend)

	// Keep reading and scheduling section requests
	for {
		select {
		case <-quit:
			return

		case section, ok := <-reqs:
			// New section retrieval requested
			if !ok {
				return
			}
			// Deduplicate retrieval requests
			unique := false

			s.lock.Lock()
			if s.responses[section] == nil {
				s.responses[section] = &response{
					done: make(chan struct{}),
				}
				unique = true
			}
			s.lock.Unlock()

			// Schedule the section for retrieval and notify the deliverer to expect this section
			if unique {
				select {
				case <-quit:
					return
				case dist <- &request{bit: s.bit, section: section}:
				}
			}
			select {
			case <-quit:
				return
			case pend <- section:
			}
		}
	}
}