// Start eager background caching of volume stats.
552712
func (s *fsResourceAnalyzer) Start() {
	s.startOnce.Do(func() {
		if s.calcPeriod <= 0 {
			klog.Info("Volume stats collection disabled.")
			return
		}
		klog.Info("Starting FS ResourceAnalyzer")
		go wait.Forever(func() { s.updateCachedPodVolumeStats() }, s.calcPeriod)
	})
}
147267
func (tc *TreeCache) Start() error {
	if !tc.state.Change(curator.LATENT, curator.STARTED) {
		return errors.New("already started")
	}
	if tc.createParentNodes {
		_, err := tc.client.Create().CreatingParentsIfNeeded().ForPath(tc.root.path)
		if err != nil {
			return fmt.Errorf("Failed to create parents: %s", err)
		}
	}
	tc.client.ConnectionStateListenable().AddListener(tc.connectionStateListener)
	if tc.client.ZookeeperClient().Connected() {
		tc.root.wasCreated()
	}
	return nil
}
1264002
func (c *FleetClient) Start(targets []string, wg *sync.WaitGroup, out, ew io.Writer) {
	// expand @* targets
	expandedTargets, err := c.expandTargets(targets)
	if err != nil {
		io.WriteString(ew, err.Error())
		return
	}

	for _, target := range expandedTargets {
		wg.Add(1)
		go doStart(c, target, wg, out, ew)
	}
	return
}
1016592
func (p *Pss) Start(srv *p2p.Server) error {
	go func() {
		ticker := time.NewTicker(defaultCleanInterval)
		cacheTicker := time.NewTicker(p.cacheTTL)
		defer ticker.Stop()
		defer cacheTicker.Stop()
		for {
			select {
			case <-cacheTicker.C:
				p.cleanFwdCache()
			case <-ticker.C:
				p.cleanKeys()
			case <-p.quitC:
				return
			}
		}
	}()
	go func() {
		for {
			select {
			case msg := <-p.outbox:
				err := p.forward(msg)
				if err != nil {
					log.Error(err.Error())
					metrics.GetOrRegisterCounter("pss.forward.err", nil).Inc(1)
				}
			case <-p.quitC:
				return
			}
		}
	}()
	log.Info("Started Pss")
	log.Info("Loaded EC keys", "pubkey", common.ToHex(crypto.FromECDSAPub(p.PublicKey())), "secp256", common.ToHex(crypto.CompressPubkey(p.PublicKey())))
	return nil
}
346947
func (p *PubSubPush) Start(acc telegraf.Accumulator) error {
	if p.MaxBodySize.Size == 0 {
		p.MaxBodySize.Size = defaultMaxBodySize
	}

	if p.ReadTimeout.Duration < time.Second {
		p.ReadTimeout.Duration = time.Second * 10
	}
	if p.WriteTimeout.Duration < time.Second {
		p.WriteTimeout.Duration = time.Second * 10
	}

	tlsConf, err := p.ServerConfig.TLSConfig()
	if err != nil {
		return err
	}

	p.server = &http.Server{
		Addr:        p.ServiceAddress,
		Handler:     http.TimeoutHandler(p, p.WriteTimeout.Duration, "timed out processing metric"),
		ReadTimeout: p.ReadTimeout.Duration,
		TLSConfig:   tlsConf,
	}

	p.ctx, p.cancel = context.WithCancel(context.Background())
	p.wg = &sync.WaitGroup{}
	p.acc = acc.WithTracking(p.MaxUndeliveredMessages)
	p.sem = make(chan struct{}, p.MaxUndeliveredMessages)
	p.undelivered = make(map[telegraf.TrackingID]chan bool)
	p.mu = &sync.Mutex{}

	p.wg.Add(1)
	go func() {
		defer p.wg.Done()
		p.receiveDelivered()
	}()

	p.wg.Add(1)
	go func() {
		defer p.wg.Done()
		if tlsConf != nil {
			p.server.ListenAndServeTLS("", "")
		} else {
			p.server.ListenAndServe()
		}
	}()

	return nil
}
340142
func (c *Discovery) Start(ctx context.Context) error {
	if c.downloader != nil {
		c.downloader.Start(ctx)
	}
	return nil
}
1084754
func (b *BackgroundEphemeralPurger) initQueue(ctx context.Context) {
	b.queueLock.Lock()
	defer b.queueLock.Unlock()

	// Create a new queue
	b.pq = newPriorityQueue()
	heap.Init(b.pq)

	allPurgeInfo, err := b.storage.GetAllPurgeInfo(ctx, b.uid)
	if err != nil {
		b.Debug(ctx, "unable to get purgeInfo: %v", allPurgeInfo)
	}
	for _, purgeInfo := range allPurgeInfo {
		if purgeInfo.IsActive {
			b.updateQueue(purgeInfo)
		}
	}
}
1097652
func (l *lcowfs) startVM() error {
	l.Lock()
	defer l.Unlock()
	if l.currentSVM != nil {
		return nil
	}

	svm, err := l.d.startServiceVMIfNotRunning(l.vmID, l.mappedDisks, fmt.Sprintf("lcowfs.startVM"))
	if err != nil {
		return err
	}

	if err = svm.createUnionMount(l.root, l.mappedDisks...); err != nil {
		return err
	}
	l.currentSVM = svm
	return nil
}
47873
func (m *MaxReplicationLagModule) Start(rateUpdateChan chan<- struct{}) {
	m.rateUpdateChan = rateUpdateChan
	m.wg.Add(1)
	go m.ProcessRecords()
}
284625
func (pm *imageProgressManager) start() {
	now := time.Now()
	pm.imageProgress.pullStart = now
	pm.lastSlowReport = now
	go func() {
		ticker := time.NewTicker(dockerImageProgressReportInterval)
		for {
			select {
			case <-ticker.C:
				msg, lastStatusTime := pm.imageProgress.get()
				t := time.Now()
				if t.Sub(lastStatusTime) > pm.activityDeadline {
					pm.inactivityFunc(pm.image, msg, lastStatusTime)
					pm.cancel()
					return
				}
				if t.Sub(pm.lastSlowReport) > pm.slowReportInterval {
					pm.slowReporter(pm.image, msg, lastStatusTime)
					pm.lastSlowReport = t
				}
				pm.reporter(pm.image, msg, lastStatusTime)
			case <-pm.stopCh:
				return
			}
		}
	}()
}
987889
func (g *Gzip) Start() error {
	switch g.Level {
	case gzip.DefaultCompression:
	case gzip.BestCompression:
	case gzip.BestSpeed:
	default:
		g.Level = gzip.DefaultCompression
	}
	return nil
}
637224
func (p *Prometheus) Start(a telegraf.Accumulator) error {
	if p.MonitorPods {
		var ctx context.Context
		ctx, p.cancel = context.WithCancel(context.Background())
		return p.start(ctx)
	}
	return nil
}
41977
func (p *Profiler) Start() error {
	if p.Dir == "" {
		if p.ProfileCPU {
			return errors.New("-profile-cpu requires -profile-output-dir to be set")
		}
		if p.ProfileHeap {
			return errors.New("-profile-heap requires -profile-output-dir to be set")
		}
	}

	if p.ProfileCPU {
		out, err := os.Create(p.generateOutPath("cpu"))
		if err != nil {
			return errors.Annotate(err, "failed to create CPU profile output file").Err()
		}
		pprof.StartCPUProfile(out)
		p.profilingCPU = true
	}

	if p.BindHTTP != "" {
		if err := p.startHTTP(); err != nil {
			return errors.Annotate(err, "failed to start HTTP server").Err()
		}
	}

	return nil
}
768333
func (f *sharedInformerFactory) WaitForCacheSync(stopCh <-chan struct{}) map[reflect.Type]bool {
	informers := func() map[reflect.Type]cache.SharedIndexInformer {
		f.lock.Lock()
		defer f.lock.Unlock()

		informers := map[reflect.Type]cache.SharedIndexInformer{}
		for informerType, informer := range f.informers {
			if f.startedInformers[informerType] {
				informers[informerType] = informer
			}
		}
		return informers
	}()

	res := map[reflect.Type]bool{}
	for informType, informer := range informers {
		res[informType] = cache.WaitForCacheSync(stopCh, informer.HasSynced)
	}
	return res
}
975716
func (vm *VM) Start() error {
	if vm.MockStart != nil {
		return vm.MockStart()
	}
	return lvm.ErrNotImplemented
}
253635
func (c *Client) AsyncStart() error {
	if err := c.Listen(); err != nil {
		return err
	}
	var started sync.WaitGroup
	started.Add(1)
	go func() {
		started.Done()
		c.Serve()
	}()
	started.Wait()
	return nil
}
1245979
func (task *UDPListenerTask) Start(wg *sync.WaitGroup) StopChan {
	return task.ExtendedStart(nil, wg)
}