// runs the main kubelet loop, closing the kubeletFinished chan when the loop exits.
// never returns.
func (ms *MinionServer) Run(hks hyperkube.Interface, _ []string) error {
	if ms.privateMountNS {
		// only the Linux version will do anything
		enterPrivateMountNamespace()
	}

	// create apiserver client
	clientConfig, err := ms.KubeletExecutorServer.CreateAPIServerClientConfig()
	if err != nil {
		// required for k8sm since we need to send api.Binding information
		// back to the apiserver
		log.Fatalf("No API client: %v", err)
	}
	ms.clientConfig = clientConfig

	// run subprocesses until ms.done is closed on return of this function
	defer close(ms.done)
	if ms.runProxy {
		go runtime.Until(ms.launchProxyServer, 5*time.Second, ms.done)
	}
	go runtime.Until(ms.launchExecutorServer, 5*time.Second, ms.done)

	// wait until minion exit is requested
	// don't close ms.exit here to avoid panics of go routines writing an error to it
	return <-ms.exit
}

func (self *procImpl) begin() runtime.Signal {
	if !self.state.transition(stateNew, stateRunning) {
		panic(fmt.Errorf("failed to transition from New to Idle state"))
	}
	defer log.V(2).Infof("started process %d", self.pid)
	var entered runtime.Latch

	// execute actions on the backlog chan
	return runtime.After(func() {
		runtime.Until(func() {
			if entered.Acquire() {
				close(self.running)
				self.wg.Add(1)
			}
			for action := range self.backlog {
				select {
				case <-self.terminate:
					return
				default:
					// signal to indicate there's room in the backlog now
					self.changed.Broadcast()
					// rely on Until to handle action panics
					action()
				}
			}
		}, self.actionHandlerCrashDelay, self.terminate)
	}).Then(func() {
		log.V(2).Infof("finished processing action backlog for process %d", self.pid)
		if !entered.Acquire() {
			self.wg.Done()
		}
	})
}

func (s *offerStorage) Init(done <-chan struct{}) {
	// zero delay, reap offers as soon as they expire
	go runtime.Until(s.ageOffers, 0, done)

	// cached offer ids for the purposes of listener notification
	idCache := &stringsCache{
		refill: func() util.StringSet {
			result := util.NewStringSet()
			for _, v := range s.offers.List() {
				if offer, ok := v.(Perishable); ok {
					result.Insert(offer.Id())
				}
			}
			return result
		},
		ttl: offerIdCacheTTL,
	}

	go runtime.Until(func() { s.notifyListeners(idCache.Strings) }, notifyListenersDelay, done)
}

// Notify runs Elect() on m, and calls Start()/Stop() on s when the
// elected master starts/stops matching 'id'. Never returns.
func Notify(m MasterElector, path, id string, s Service, abort <-chan struct{}) {
	n := &notifier{id: Master(id), service: s}
	n.changed = make(chan struct{})
	finished := runtime.After(func() {
		runtime.Until(func() {
			for {
				w := m.Elect(path, id)
				for {
					select {
					case <-abort:
						return
					case event, open := <-w.ResultChan():
						if !open {
							break
						}
						if event.Type != watch.Modified {
							continue
						}
						electedMaster, ok := event.Object.(Master)
						if !ok {
							glog.Errorf("Unexpected object from election channel: %v", event.Object)
							break
						}

						n.lock.Lock()
						n.desired = electedMaster
						n.lock.Unlock()

						// notify serviceLoop, but don't block. If a change
						// is queued already it will see the new n.desired.
						select {
						case n.changed <- struct{}{}:
						}
					}
				}
			}
		}, 0, abort)
	})
	runtime.Until(func() { n.serviceLoop(finished) }, 0, abort)
}

// Notify runs Elect() on m, and calls Start()/Stop() on s when the
// elected master starts/stops matching 'id'. Never returns.
func Notify(m MasterElector, path, id string, s Service, abort <-chan struct{}) {
	n := &notifier{id: Master(id), service: s}
	n.cond = sync.NewCond(&n.lock)
	finished := runtime.After(func() {
		runtime.Until(func() {
			for {
				w := m.Elect(path, id)
				for {
					select {
					case <-abort:
						return
					case event, open := <-w.ResultChan():
						if !open {
							break
						}
						if event.Type != watch.Modified {
							continue
						}
						electedMaster, ok := event.Object.(Master)
						if !ok {
							glog.Errorf("Unexpected object from election channel: %v", event.Object)
							break
						}
						func() {
							n.lock.Lock()
							defer n.lock.Unlock()
							n.desired = electedMaster
							if n.desired != n.current {
								n.cond.Signal()
							}
						}()
					}
				}
			}
		}, 0, abort)
	})
	runtime.Until(func() { n.serviceLoop(finished) }, 0, abort)
}

// currently monitors for "pod deleted" events, upon which handle()
// is invoked.
func (k *deleter) Run(updates <-chan queue.Entry, done <-chan struct{}) {
	go runtime.Until(func() {
		for {
			entry := <-updates
			pod := entry.Value().(*Pod)
			if entry.Is(queue.DELETE_EVENT) {
				if err := k.deleteOne(pod); err != nil {
					log.Error(err)
				}
			} else if !entry.Is(queue.POP_EVENT) {
				k.qr.updatesAvailable()
			}
		}
	}, 1*time.Second, done)
}

// perform one-time initialization actions upon the first registration event received from Mesos.
func (k *KubernetesScheduler) onInitialRegistration(driver bindings.SchedulerDriver) {
	defer close(k.registration)

	if k.failoverTimeout > 0 {
		refreshInterval := k.schedcfg.FrameworkIdRefreshInterval.Duration
		if k.failoverTimeout < k.schedcfg.FrameworkIdRefreshInterval.Duration.Seconds() {
			refreshInterval = time.Duration(math.Max(1, k.failoverTimeout/2)) * time.Second
		}
		go runtime.Until(k.storeFrameworkId, refreshInterval, k.terminate)
	}

	r1 := k.makeTaskRegistryReconciler()
	r2 := k.makePodRegistryReconciler()

	k.reconciler = newReconciler(k.asRegisteredMaster, k.makeCompositeReconciler(r1, r2),
		k.reconcileCooldown, k.schedcfg.ExplicitReconciliationAbortTimeout.Duration, k.terminate)
	go k.reconciler.Run(driver)

	if k.reconcileInterval > 0 {
		ri := time.Duration(k.reconcileInterval) * time.Second
		time.AfterFunc(k.schedcfg.InitialImplicitReconciliationDelay.Duration, func() { runtime.Until(k.reconciler.RequestImplicit, ri, k.terminate) })
		log.Infof("will perform implicit task reconciliation at interval: %v after %v", ri, k.schedcfg.InitialImplicitReconciliationDelay.Duration)
	}
}

func (kl *kubeletExecutor) ListenAndServe(address net.IP, port uint, tlsOptions *kubelet.TLSOptions, enableDebuggingHandlers bool) {
	// this func could be called many times, depending how often the HTTP server crashes,
	// so only execute certain initialization procs once
	kl.initialize.Do(func() {
		if kl.runProxy {
			go runtime.Until(kl.runProxyService, 5*time.Second, kl.executorDone)
		}
		go func() {
			if _, err := kl.driver.Run(); err != nil {
				log.Fatalf("executor driver failed: %v", err)
			}
			log.Info("executor Run completed")
		}()
	})
	log.Infof("Starting kubelet server...")
	kubelet.ListenAndServeKubeletServer(kl, address, port, tlsOptions, enableDebuggingHandlers)
}

func (s *SchedulerServer) Run(hks hyperkube.Interface, _ []string) error {
	// get scheduler low-level config
	sc := schedcfg.CreateDefaultConfig()
	if s.SchedulerConfigFileName != "" {
		f, err := os.Open(s.SchedulerConfigFileName)
		if err != nil {
			log.Fatalf("Cannot open scheduler config file: %v", err)
		}

		err = sc.Read(bufio.NewReader(f))
		if err != nil {
			log.Fatalf("Invalid scheduler config file: %v", err)
		}
	}

	schedulerProcess, driverFactory, etcdClient, eid := s.bootstrap(hks, sc)

	if s.EnableProfiling {
		profile.InstallHandler(s.mux)
	}
	go runtime.Until(func() {
		log.V(1).Info("Starting HTTP interface")
		log.Error(http.ListenAndServe(net.JoinHostPort(s.Address.String(), strconv.Itoa(s.Port)), s.mux))
	}, sc.HttpBindInterval.Duration, schedulerProcess.Terminal())

	if s.HA {
		validation := ha.ValidationFunc(validateLeadershipTransition)
		srv := ha.NewCandidate(schedulerProcess, driverFactory, validation)
		path := fmt.Sprintf(meta.DefaultElectionFormat, s.FrameworkName)
		sid := uid.New(eid.Group(), "").String()
		log.Infof("registering for election at %v with id %v", path, sid)
		go election.Notify(election.NewEtcdMasterElector(etcdClient), path, sid, srv, nil)
	} else {
		log.Infoln("self-electing in non-HA mode")
		schedulerProcess.Elect(driverFactory)
	}
	return s.awaitFailover(schedulerProcess, func() error { return s.failover(s.getDriver(), hks) })
}