// NewStartWorkerArgs initializes StartWorkerArgs with the required arguments.
7155
func NewStartWorkerArgs(scopeURL string) *StartWorkerArgs {
	args := new(StartWorkerArgs)
	args.ScopeURL = scopeURL
	return args
}
136998
func (wm *WorkerManager) Start() {
	go wm.processBatch()
	go wm.commitBatch()
	for {
		startIdle := time.Now()
		// force manager stop to be checked first
		select {
		case <-wm.managerStop:
			return
		default:
			select {
			case batch := <-wm.inputChannel:
				{
					wm.metrics.wMsIdle().Update(time.Since(startIdle))
					if Logger.IsAllowed(TraceLevel) {
						Trace(wm, "WorkerManager got batch")
					}
					wm.metrics.wMsBatchDuration().Time(func() {
						wm.startBatch(batch)
					})
					if Logger.IsAllowed(TraceLevel) {
						Trace(wm, "WorkerManager got batch processed")
					}
				}
			case <-wm.managerStop:
				return
			}
		}
	}
}
164914
func NewStringsWorker(config StringsConfig) (*StringsWorker, error) {
	if err := config.Validate(); err != nil {
		return nil, errors.Trace(err)
	}
	sw := &StringsWorker{
		config: config,
	}
	err := catacomb.Invoke(catacomb.Plan{
		Site: &sw.catacomb,
		Work: sw.loop,
	})
	if err != nil {
		return nil, errors.Trace(err)
	}
	return sw, nil
}
1001075
func (w *Worker) Start() {
	handlerInterrupted := false
	go func() {
		for taskAndStrategy := range w.HandlerInputChannel {
			result := taskAndStrategy.Strategy(w, taskAndStrategy.WorkerTask.Msg, taskAndStrategy.WorkerTask.Id())
		Loop:
			for !handlerInterrupted {
				timeout := time.NewTimer(5 * time.Second)
				select {
				case w.HandlerOutputChannel <- result:
					timeout.Stop()
					break Loop
				case <-timeout.C:
				}
			}
			handlerInterrupted = false
		}
	}()

	go func() {
		for taskAndStrategy := range w.InputChannel {
			taskAndStrategy.WorkerTask.Callee = w
			w.HandlerInputChannel <- taskAndStrategy
			timeout := time.NewTimer(w.TaskTimeout)
			select {
			case result := <-w.HandlerOutputChannel:
				{
					w.OutputChannel <- result
				}
			case <-timeout.C:
				{
					handlerInterrupted = true
					w.OutputChannel <- &TimedOutResult{taskAndStrategy.WorkerTask.Id()}
				}
			}
			timeout.Stop()
		}
	}()
}
1369033
func runWorkers(
	ctx context.Context,
	o *gcs.Object,
	bucket gcs.Bucket) (results []result, err error) {
	b := syncutil.NewBundle(ctx)

	// Set up a channel that is closed upon SIGINT.
	stop := make(chan struct{})
	go func() {
		c := make(chan os.Signal, 1)
		signal.Notify(c, os.Interrupt)

		<-c
		log.Printf("SIGINT received. Stopping soon...")
		close(stop)
	}()

	// Start several workers making random reads.
	var wg sync.WaitGroup
	resultChan := make(chan result)
	for i := 0; i < *fWorkers; i++ {
		wg.Add(1)
		b.Add(func(ctx context.Context) (err error) {
			defer wg.Done()
			err = makeReads(ctx, o, bucket, resultChan, stop)
			if err != nil {
				err = fmt.Errorf("makeReads: %v", err)
				return
			}

			return
		})
	}

	go func() {
		wg.Wait()
		close(resultChan)
	}()

	// Accumulate results.
	b.Add(func(ctx context.Context) (err error) {
		for r := range resultChan {
			results = append(results, r)
		}

		return
	})

	err = b.Join()
	return
}
387842
func (w *DefaultWorker) Start() (<-chan *Task, <-chan error) {
	w.smu.Lock()
	defer w.smu.Unlock()

	w.state = open
	w.lifecycle.SetQueues(w.availableTasks, w.workingTasks)
	w.janitorRunner = newJanitorRunner(w.pool, w.detector, w.janitor, w.availableTasks)

	errs1 := w.janitorRunner.Start()
	tasks, errs2 := w.lifecycle.Listen()

	return tasks, concatErrs(errs1, errs2)
}
859509
func NewStarterClient(cheStarterURL, openshiftMasterURL string, namespace string, client *http.Client) Client {
	return &StarterClient{cheStarterURL: cheStarterURL, openshiftMasterURL: openshiftMasterURL, namespace: namespace, client: client}
}
161077
func startWS(ctx *cli.Context) (err error) {
	server, cleanup, err := newServer(ctx)
	if err != nil {
		return err
	}
	defer cleanup()

	listener, err := net.Listen("tcp", ctx.String("addr"))
	if err != nil {
		return err
	}
	origins := ctx.StringSlice("origins")
	log.Info("websocket", "address", listener.Addr().String(), "origins", origins)

	return http.Serve(listener, server.WebsocketHandler(origins))
}
535501
func (c *ReportingClient) RunStepStarted(ctx context.Context, args RunStepStartedArgs) error {
	return c.postJSON("runstepstarted", args)
}
616802
func newSplitCloneWorker(wr *wrangler.Wrangler, cell, keyspace, shard string, online, offline bool, excludeTables []string, chunkCount, minRowsPerChunk, sourceReaderCount, writeQueryMaxRows, writeQueryMaxSize, destinationWriterCount, minHealthyTablets int, tabletType topodatapb.TabletType, maxTPS, maxReplicationLag int64, useConsistentSnapshot bool) (Worker, error) {
	return newCloneWorker(wr, horizontalResharding, cell, keyspace, shard, online, offline, nil /* tables */, excludeTables, chunkCount, minRowsPerChunk, sourceReaderCount, writeQueryMaxRows, writeQueryMaxSize, destinationWriterCount, minHealthyTablets, tabletType, maxTPS, maxReplicationLag, useConsistentSnapshot)
}
257672
func LoadStart(listener func(*vecty.Event)) *vecty.EventListener {
	return &vecty.EventListener{Name: "loadstart", Listener: listener}
}
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)
}
669195
func StartWorkers(ctx context.Context, rnr *Runner, tasks <-chan task.Request) {
	var wg sync.WaitGroup
	defer wg.Wait()
	var hcmgr htfnmgr

	for {
		select {
		case <-ctx.Done():
			return
		case task := <-tasks:
			p := hcmgr.getPipe(ctx, rnr, task.Config)
			if p == nil {
				wg.Add(1)
				go runTaskReq(rnr, &wg, task)
				continue
			}

			rnr.Start()
			select {
			case <-ctx.Done():
				return
			case p <- task:
				rnr.Complete()
			}
		}
	}
}
744442
func NewStartWithBigInt(now time.Time, mod *big.Int, nPlayers int) Start {
	ret := newStart(now, nPlayers)
	ret.Params = NewFlipParametersWithBig(mod.Bytes())
	return ret
}
1357542
func (cc *CeleryClient) StartWorkerWithContext(ctx context.Context) {
	cc.worker.StartWorkerWithContext(ctx)
}
34528
func (c *Crawler) launchWorker(ctx *URLContext) *worker {
	// Initialize index and channels
	i := len(c.workers) + 1
	pop := newPopChannel()

	// Create the worker
	w := &worker{
		host:    ctx.normalizedURL.Host,
		index:   i,
		push:    c.push,
		pop:     pop,
		stop:    c.stop,
		enqueue: c.enqueue,
		wg:      c.wg,
		logFunc: getLogFunc(c.Options.Extender, c.Options.LogFlags, i),
		opts:    c.Options,
	}

	// Increment wait group count
	c.wg.Add(1)

	// Launch worker
	go w.run()
	c.logFunc(LogInfo, "worker %d launched for host %s", i, w.host)
	c.workers[w.host] = w

	return w
}
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
}