// StartWorker -
// scopeURL -
483911
func (c *ServiceWorker) StartWorker(scopeURL string) (*gcdmessage.ChromeResponse, error) {
	var v ServiceWorkerStartWorkerParams
	v.ScopeURL = scopeURL
	return c.StartWorkerWithParams(&v)
}
1171325
func (nm *NamespaceController) worker() {
	workFunc := func() bool {
		key, quit := nm.queue.Get()
		if quit {
			return true
		}
		defer nm.queue.Done(key)

		err := nm.syncNamespaceFromKey(key.(string))
		if err == nil {
			// no error, forget this entry and return
			nm.queue.Forget(key)
			return false
		}

		if estimate, ok := err.(*deletion.ResourcesRemainingError); ok {
			t := estimate.Estimate/2 + 1
			klog.V(4).Infof("Content remaining in namespace %s, waiting %d seconds", key, t)
			nm.queue.AddAfter(key, time.Duration(t)*time.Second)
		} else {
			// rather than wait for a full resync, re-add the namespace to the queue to be processed
			nm.queue.AddRateLimited(key)
			utilruntime.HandleError(err)
		}
		return false
	}

	for {
		quit := workFunc()

		if quit {
			return
		}
	}
}
1117834
func NewWorker(config WorkerConfig) (worker.Worker, error) {
	if err := config.Validate(); err != nil {
		return nil, errors.Trace(err)
	}

	started := make(chan struct{})
	w := &agentConfigUpdater{
		config:       config,
		mongoProfile: config.MongoProfile,
	}
	w.tomb.Go(func() error {
		return w.loop(started)
	})
	select {
	case <-started:
	case <-time.After(10 * time.Second):
		return nil, errors.New("worker failed to start properly")
	}
	return w, nil
}
614249
func manifoldStart(apiCaller base.APICaller) (worker.Worker, error) {
	client, err := metricsmanager.NewClient(apiCaller)
	if err != nil {
		return nil, errors.Trace(err)
	}
	w, err := newMetricsManager(client, nil)
	if err != nil {
		return nil, errors.Trace(err)
	}
	return w, nil
}
387869
func (a adminAPIHandlers) StartProfilingHandler(w http.ResponseWriter, r *http.Request) {
	ctx := newContext(r, w, "StartProfiling")

	objectAPI := validateAdminReq(ctx, w, r)
	if objectAPI == nil {
		return
	}

	vars := mux.Vars(r)
	profiler := vars["profilerType"]

	thisAddr, err := xnet.ParseHost(GetLocalPeer(globalEndpoints))
	if err != nil {
		writeErrorResponseJSON(ctx, w, toAdminAPIErr(ctx, err), r.URL)
		return
	}

	// Start profiling on remote servers.
	hostErrs := globalNotificationSys.StartProfiling(profiler)

	// Start profiling locally as well.
	{
		if globalProfiler != nil {
			globalProfiler.Stop()
		}
		prof, err := startProfiler(profiler, "")
		if err != nil {
			hostErrs = append(hostErrs, NotificationPeerErr{
				Host: *thisAddr,
				Err:  err,
			})
		} else {
			globalProfiler = prof
			hostErrs = append(hostErrs, NotificationPeerErr{
				Host: *thisAddr,
			})
		}
	}

	var startProfilingResult []StartProfilingResult

	for _, nerr := range hostErrs {
		result := StartProfilingResult{NodeName: nerr.Host.String()}
		if nerr.Err != nil {
			result.Error = nerr.Err.Error()
		} else {
			result.Success = true
		}
		startProfilingResult = append(startProfilingResult, result)
	}

	// Create JSON result and send it to the client
	startProfilingResultInBytes, err := json.Marshal(startProfilingResult)
	if err != nil {
		writeErrorResponseJSON(ctx, w, toAdminAPIErr(ctx, err), r.URL)
		return
	}

	writeSuccessResponseJSON(w, []byte(startProfilingResultInBytes))
}
944949
func (mj *mirrorJob) startMirror(ctx context.Context, cancelMirror context.CancelFunc) {
	var totalBytes int64
	var totalObjects int64

	stopParallel := func() {
		close(mj.queueCh)
		mj.parallel.wait()
	}

	URLsCh := prepareMirrorURLs(mj.sourceURL, mj.targetURL, mj.isFake, mj.isOverwrite, mj.isRemove, mj.excludeOptions, mj.encKeyDB)

	for {
		select {
		case sURLs, ok := <-URLsCh:
			if !ok {
				stopParallel()
				return
			}
			if sURLs.Error != nil {
				stopParallel()
				mj.statusCh <- sURLs
				return
			}

			if sURLs.SourceContent != nil {
				if mj.olderThan != "" && isOlder(sURLs.SourceContent.Time, mj.olderThan) {
					continue
				}
				if mj.newerThan != "" && isNewer(sURLs.SourceContent.Time, mj.newerThan) {
					continue
				}
				// copy
				totalBytes += sURLs.SourceContent.Size
			}

			totalObjects++
			mj.TotalBytes = totalBytes
			mj.TotalObjects = totalObjects
			mj.status.SetTotal(totalBytes)

			// Save total count.
			sURLs.TotalCount = mj.TotalObjects
			// Save totalSize.
			sURLs.TotalSize = mj.TotalBytes

			if sURLs.SourceContent != nil {
				mj.queueCh <- func() URLs {
					return mj.doMirror(ctx, cancelMirror, sURLs)
				}
			} else if sURLs.TargetContent != nil && mj.isRemove {
				mj.queueCh <- func() URLs {
					return mj.doRemove(sURLs)
				}
			}
		case <-mj.trapCh:
			stopParallel()
			cancelMirror()
			return
		}
	}
}
1101775
func (c *Client) workHandler() {
	for {
		select {
		case w := <-c.work:
			c.process(w)
		case <-time.After(c.keepAlive):
			c.process(keepAliveData)
		case <-c.disconnect:
			return
		}
	}
}
503448
func CreateWorker(args *WorkArgs, createdAt time.Time, fireAt time.Time, f func(args *WorkArgs) error) *TimedWorker {
	delay := fireAt.Sub(createdAt)
	if delay <= 0 {
		go f(args)
		return nil
	}
	timer := time.AfterFunc(delay, func() { f(args) })
	return &TimedWorker{
		WorkItem:  args,
		CreatedAt: createdAt,
		FireAt:    fireAt,
		Timer:     timer,
	}
}
608120
func (w *workManager) Start() {

	w.mutex.Lock()
	defer w.mutex.Unlock()

	if w.state == workManagerStopped {
		w.state = workManagerRunning
		go func() {
			for {
				select {
				case <-w.collectq.Err:
					//TODO: log error
				case <-w.processq.Err:
					//TODO: log error
				case <-w.publishq.Err:
					//TODO: log error
				case <-w.kill:
					return
				}
			}
		}()
	}
}
1305907
func (p *provisioner) saveApplicationWorker(appName string, aw *applicationWorker) {
	p.mu.Lock()
	defer p.mu.Unlock()

	if p.appWorkers == nil {
		p.appWorkers = make(map[string]*applicationWorker)
	}
	p.appWorkers[appName] = aw
}
1286432
func StartWorker(config *config.Config) error {
	var err error
	app.worker, err = worker.NewWorker(config)
	if err != nil {
		return err
	}

	app.worker.Start()

	return nil
}
1143807
func (remote *RemoteDebugger) StartProfiler() error {
	_, err := remote.SendRequest("Profiler.start", nil)
	return err
}
1005413
func start(c *cli.Context) (err error) {
	// set legacy watcher
	if c.Bool("legacy") {
		r.Settings.Legacy.Set(c.Bool("legacy"), 1)
	}
	// set server
	if c.Bool("server") {
		r.Server.Set(c.Bool("server"), c.Bool("open"), realize.Port, realize.Host)
	}

	// check no-config and read
	if !c.Bool("no-config") {
		// read a config if exist
		r.Settings.Read(&r)
		if c.String("name") != "" {
			// filter by name flag if exist
			r.Schema.Projects = r.Schema.Filter("Name", c.String("name"))
		}
		// increase file limit
		if r.Settings.FileLimit != 0 {
			if err = r.Settings.Flimit(); err != nil {
				return err
			}
		}

	}
	// check project list length
	if len(r.Schema.Projects) == 0 {
		// create a new project based on given params
		project := r.Schema.New(c)
		// Add to projects list
		r.Schema.Add(project)
		// save config
		if !c.Bool("no-config") {
			err = r.Settings.Write(r)
			if err != nil {
				return err
			}
		}
	}
	// Start web server
	if r.Server.Status {
		r.Server.Parent = &r
		err = r.Server.Start()
		if err != nil {
			return err
		}
		err = r.Server.OpenURL()
		if err != nil {
			return err
		}
	}
	// start workflow
	return r.Start()
}
1299000
func NewWorker(opt WorkerOpt) (*Worker, error) {
	imageRefChecker := imagerefchecker.New(imagerefchecker.Opt{
		ImageStore:   opt.ImageStore,
		Snapshotter:  opt.Snapshotter,
		ContentStore: opt.ContentStore,
	})

	cm, err := cache.NewManager(cache.ManagerOpt{
		Snapshotter:     opt.Snapshotter,
		MetadataStore:   opt.MetadataStore,
		PruneRefChecker: imageRefChecker,
	})
	if err != nil {
		return nil, err
	}

	sm, err := source.NewManager()
	if err != nil {
		return nil, err
	}

	is, err := containerimage.NewSource(containerimage.SourceOpt{
		Snapshotter:   opt.Snapshotter,
		ContentStore:  opt.ContentStore,
		Applier:       opt.Applier,
		ImageStore:    opt.ImageStore,
		CacheAccessor: cm,
		ResolverOpt:   opt.ResolveOptionsFunc,
	})
	if err != nil {
		return nil, err
	}

	sm.Register(is)

	if err := git.Supported(); err == nil {
		gs, err := git.NewSource(git.Opt{
			CacheAccessor: cm,
			MetadataStore: opt.MetadataStore,
		})
		if err != nil {
			return nil, err
		}
		sm.Register(gs)
	} else {
		logrus.Warnf("git source cannot be enabled: %v", err)
	}

	hs, err := http.NewSource(http.Opt{
		CacheAccessor: cm,
		MetadataStore: opt.MetadataStore,
	})
	if err != nil {
		return nil, err
	}

	sm.Register(hs)

	ss, err := local.NewSource(local.Opt{
		CacheAccessor: cm,
		MetadataStore: opt.MetadataStore,
	})
	if err != nil {
		return nil, err
	}
	sm.Register(ss)

	iw, err := imageexporter.NewImageWriter(imageexporter.WriterOpt{
		Snapshotter:  opt.Snapshotter,
		ContentStore: opt.ContentStore,
		Applier:      opt.Applier,
		Differ:       opt.Differ,
	})
	if err != nil {
		return nil, err
	}

	return &Worker{
		WorkerOpt:     opt,
		CacheManager:  cm,
		SourceManager: sm,
		imageWriter:   iw,
		ImageSource:   is,
	}, nil
}
1369637
func (wi *Instance) setAndStartWorker(ctx context.Context, wrk Worker, wr *wrangler.Wrangler) (chan struct{}, error) {
	wi.currentWorkerMutex.Lock()
	defer wi.currentWorkerMutex.Unlock()

	if wi.currentContext != nil {
		return nil, vterrors.Errorf(vtrpcpb.Code_UNAVAILABLE, "A worker job is already in progress: %v", wi.currentWorker.StatusAsText())
	}

	if wi.currentWorker != nil {
		// During the grace period, we answer with a retryable error.
		// This way any automation can retry to issue 'Reset' and then the original
		// command. We can end up in this situation when the automation job was
		// restarted and therefore the previously running vtworker command was
		// canceled.
		// TODO(mberlin): This can be simplified when we move to a model where
		// vtworker runs commands independent of an RPC and the automation polls for
		// the current status, based on an assigned unique id, instead.
		const gracePeriod = 1 * time.Minute
		sinceLastStop := time.Since(wi.lastRunStopTime)
		if sinceLastStop <= gracePeriod {
			return nil, vterrors.Errorf(vtrpcpb.Code_UNAVAILABLE,
				"A worker job was recently stopped (%f seconds ago): If you run commands manually, run the 'Reset' command to clear the vtworker state. Job: %v",
				sinceLastStop.Seconds(),
				wi.currentWorker)
		}

		// We return FAILED_PRECONDITION to signal that a manual resolution is required.
		return nil, vterrors.Errorf(vtrpcpb.Code_FAILED_PRECONDITION,
			"The worker job was stopped %.1f minutes ago, but not reset. Run the 'Reset' command to clear it manually. Job: %v",
			time.Since(wi.lastRunStopTime).Minutes(),
			wi.currentWorker)
	}

	wi.currentWorker = wrk
	wi.currentMemoryLogger = logutil.NewMemoryLogger()
	wi.currentContext, wi.currentCancelFunc = context.WithCancel(ctx)
	wi.lastRunError = nil
	wi.lastRunStopTime = time.Unix(0, 0)
	done := make(chan struct{})
	wranglerLogger := wr.Logger()
	if wr == wi.wr {
		// If it's the default wrangler, do not reuse its logger because it may have been set before.
		// Resuing it would result into an endless recursion.
		wranglerLogger = logutil.NewConsoleLogger()
	}
	wr.SetLogger(logutil.NewTeeLogger(wi.currentMemoryLogger, wranglerLogger))

	// one go function runs the worker, changes state when done
	go func() {
		log.Infof("Starting worker...")
		span, ctx := trace.NewSpan(wi.currentContext, "work")
		span.Annotate("extra", wrk.State().String())
		defer span.Finish()
		var err error

		// Catch all panics and always save the execution state at the end.
		defer func() {
			// The recovery code is a copy of servenv.HandlePanic().
			if x := recover(); x != nil {
				log.Errorf("uncaught vtworker panic: %v\n%s", x, tb.Stack(4))
				err = vterrors.Errorf(vtrpcpb.Code_INTERNAL, "uncaught vtworker panic: %v", x)
			}

			wi.currentWorkerMutex.Lock()
			wi.currentContext = nil
			wi.currentCancelFunc = nil
			wi.lastRunError = err
			wi.lastRunStopTime = time.Now()
			wi.currentWorkerMutex.Unlock()
			close(done)
		}()

		// run will take a long time
		err = wrk.Run(ctx)

		// If the context was canceled, include the respective error c ode.
		select {
		case <-ctx.Done():
			// Context is done i.e. probably canceled.
			if wi.currentContext.Err() == context.Canceled {
				err = vterrors.Errorf(vtrpcpb.Code_CANCELED, "vtworker command was canceled: %v", err)
			}
		default:
		}
	}()

	return done, nil
}
611997
func (config ManifoldConfig) start(context dependency.Context) (worker.Worker, error) {
	if err := config.Validate(); err != nil {
		return nil, errors.Trace(err)
	}

	var stTracker workerstate.StateTracker
	if err := context.Get(config.StateName, &stTracker); err != nil {
		return nil, errors.Trace(err)
	}

	pool, err := stTracker.Use()
	if err != nil {
		return nil, errors.Trace(err)
	}

	w, err := config.NewWorker(Config{
		Logger:               config.Logger,
		WatcherFactory:       func() BackingWatcher { return pool.SystemState().WatchAllModels(pool) },
		PrometheusRegisterer: config.PrometheusRegisterer,
		Cleanup:              func() { _ = stTracker.Done() },
	})
	if err != nil {
		_ = stTracker.Done()
		return nil, errors.Trace(err)
	}
	return w, nil
}
285446
func newBlockRetrievalWorker(bg blockGetter, q *blockRetrievalQueue,
	workCh channels.Channel) *blockRetrievalWorker {
	brw := &blockRetrievalWorker{
		blockGetter: bg,
		stopCh:      make(chan struct{}),
		doneCh:      make(chan struct{}),
		queue:       q,
		workCh:      workCh,
	}
	go brw.run()
	return brw
}