// Runs the job.
// If some of the mapper failed, one of the error is returned.
482342
func (job *MapOnlyJob) Run() error {
	if job.NewMapperF == nil {
		return errors.New("MapOnlyJob: NewMapperF undefined!")
	}
	if job.Source == nil {
		return errors.New("MapOnlyJob: Source undefined!")
	}
	totalPart := 0
	endss := make([][]chan error, 0, len(job.Source))
	for i := range job.Source {
		partCount, err := job.Source[i].PartCount()
		if err != nil {
			return err
		}
		ends := make([]chan error, 0, partCount)
		for part := 0; part < partCount; part++ {
			end := make(chan error, 1)
			ends = append(ends, end)
			go func(i, part, totalPart int, end chan error) {
				end <- func() error {
					mapper := job.NewMapperF(i, part)
					key, val := mapper.NewKey(), mapper.NewVal()
					cs := make([]sophie.Collector, 0, len(job.Dest))
					for _, dst := range job.Dest {
						c, err := dst.Collector(totalPart)
						if err != nil {
							return errorsp.WithStacksAndMessage(err, "open collector for source %d part %d failed", i, part)
						}
						defer c.Close()
						cs = append(cs, c)
					}
					iter, err := job.Source[i].Iterator(part)
					if err != nil {
						return errorsp.WithStacksAndMessage(err, " open source %d part %d failed", i, part)
					}
					defer iter.Close()

					for {
						if err := iter.Next(key, val); err != nil {
							if errorsp.Cause(err) != io.EOF {
								return errorsp.WithStacksAndMessage(err, "next failed")
							}
							break
						}
						if err := mapper.Map(key, val, cs); err != nil {
							if errorsp.Cause(err) == EOM {
								log.Print("EOM returned, exit early")
								break
							}
							return errorsp.WithStacksAndMessage(err, "mapping %v %v failed", key, val)
						}
					}
					return errorsp.WithStacksAndMessage(mapper.MapEnd(cs), "map end failed")
				}()
			}(i, part, totalPart, end)
			totalPart++
		}
		endss = append(endss, ends)
	}
	var errReturned error
	for _, ends := range endss {
		for part, end := range ends {
			log.Printf("Waiting for mapper %d...", part)
			if err := <-end; err != nil {
				log.Printf("Error returned for part %d: %v", part, err)
				errReturned = err
			}
			log.Printf("No error for mapper %d...", part)
		}
	}
	return errReturned
}
1001192
func (m *Menu) Run() (map[flux.ResourceID][]ContainerUpdate, error) {
	specs := make(map[flux.ResourceID][]ContainerUpdate)
	if m.selectable == 0 {
		return specs, errors.New("No changes found.")
	}

	m.printInteractive()
	m.wr.hideCursor()
	defer m.wr.showCursor()

	for {
		ascii, keyCode, err := getChar()
		if err != nil {
			return specs, err
		}

		switch ascii {
		case 3, 27, 'q':
			return specs, errors.New("Aborted.")
		case ' ':
			m.toggleSelected()
		case 13:
			for _, item := range m.items {
				if item.checked {
					specs[item.id] = append(specs[item.id], item.update)
				}
			}
			m.wr.writeln("")
			return specs, nil
		case 9, 'j':
			m.cursorDown()
		case 'k':
			m.cursorUp()
		default:
			switch keyCode {
			case 40:
				m.cursorDown()
			case 38:
				m.cursorUp()
			}
		}

	}
}
837188
func (sdw *SplitDiffWorker) Run(ctx context.Context) error {
	resetVars()
	err := sdw.run(ctx)

	sdw.SetState(WorkerStateCleanUp)
	cerr := sdw.cleaner.CleanUp(sdw.wr)
	if cerr != nil {
		if err != nil {
			sdw.wr.Logger().Errorf2(cerr, "CleanUp failed in addition to job error")
		} else {
			err = cerr
		}
	}
	if err != nil {
		sdw.wr.Logger().Errorf2(err, "Run() error")
		sdw.SetState(WorkerStateError)
		return err
	}
	sdw.SetState(WorkerStateDone)
	return nil
}
79997
func Run(driverName string, args []string, drivers map[string]StoreGenerator, chanSize, bulkTripleOpSize, builderSize int, rl repl.ReadLiner) int {
	driver, err := InitializeDriver(driverName, drivers)
	if err != nil {
		fmt.Fprintln(os.Stderr, err)
		return 2
	}
	return Eval(context.Background(), args, InitializeCommands(driver, chanSize, bulkTripleOpSize, builderSize, rl, make(chan bool)))
}
831729
func (runner *gleamRunner) runMapperReducer() {
	ctx, cancel := context.WithCancel(context.Background())
	defer cancel()

	if runner.Option.IsProfiling {
		cpuProfFile := fmt.Sprintf("mr_cpu_%d-s%d-t%d.pprof", runner.Option.HashCode,
			runner.Option.StepId, runner.Option.TaskId)
		pwd, _ := os.Getwd()
		// println("saving pprof to", pwd+"/"+cpuProfFile)

		cf, err := os.Create(cpuProfFile)
		if err != nil {
			log.Fatalf("failed to create cpu profile file %s: %v", pwd+"/"+cpuProfFile, err)
		}
		pprof.StartCPUProfile(cf)
		defer pprof.StopCPUProfile()

		memProfFile := fmt.Sprintf("mr_mem_%d-s%d-t%d.pprof", runner.Option.HashCode,
			runner.Option.StepId, runner.Option.TaskId)

		mf, err := os.Create(memProfFile)
		// println("saving pprof to", pwd+"/"+memProfFile)
		if err != nil {
			log.Fatalf("failed to create memory profile file %s: %v", pwd+"/"+memProfFile, err)
		}

		defer func() {
			runtime.GC()
			pprof.Lookup("heap").WriteTo(mf, 0)

		}()
	}

	stat.FlowHashCode = uint32(runner.Option.HashCode)
	stat.Stats = []*pb.InstructionStat{
		{
			StepId: int32(runner.Option.StepId),
			TaskId: int32(runner.Option.TaskId),
		},
	}

	if runner.Option.Mapper != "" {
		if fn, ok := mappers[MapperId(runner.Option.Mapper)]; ok {
			if err := runner.processMapper(ctx, fn.Mapper); err != nil {
				log.Fatalf("Failed to execute mapper %v: %v", os.Args, err)
			}
			return
		}
		log.Fatalf("Missing mapper function %v. Args: %v", runner.Option.Mapper, os.Args)
	}

	if runner.Option.Reducer != "" {
		if runner.Option.KeyFields == "" {
			log.Fatalf("Also expecting values for -gleam.keyFields! Actual arguments: %v", os.Args)
		}
		if fn, ok := reducers[ReducerId(runner.Option.Reducer)]; ok {
			keyPositions := strings.Split(runner.Option.KeyFields, ",")
			var keyIndexes []int
			for _, keyPosition := range keyPositions {
				keyIndex, keyIndexError := strconv.Atoi(keyPosition)
				if keyIndexError != nil {
					log.Fatalf("Failed to parse key index positions %v: %v", runner.Option.KeyFields, keyIndexError)
				}
				keyIndexes = append(keyIndexes, keyIndex)
			}

			if err := runner.processReducer(ctx, fn.Reducer, keyIndexes); err != nil {
				log.Fatalf("Failed to execute reducer %v: %v", os.Args, err)
			}
			return
		}
		log.Fatalf("Missing reducer function %v. Args: %v", runner.Option.Reducer, os.Args)

	}

	log.Fatalf("Failed to find function to execute. Args: %v", os.Args)
}
643708
func (vsdw *VerticalSplitDiffWorker) Run(ctx context.Context) error {
	resetVars()
	err := vsdw.run(ctx)

	vsdw.SetState(WorkerStateCleanUp)
	cerr := vsdw.cleaner.CleanUp(vsdw.wr)
	if cerr != nil {
		if err != nil {
			vsdw.wr.Logger().Errorf2(cerr, "CleanUp failed in addition to job error")
		} else {
			err = cerr
		}
	}
	if err != nil {
		vsdw.SetState(WorkerStateError)
		return err
	}
	vsdw.SetState(WorkerStateDone)
	return nil
}
391354
func (jm *JobController) Run(workers int, stopCh <-chan struct{}) {
	defer utilruntime.HandleCrash()
	defer jm.queue.ShutDown()

	klog.Infof("Starting job controller")
	defer klog.Infof("Shutting down job controller")

	if !controller.WaitForCacheSync("job", stopCh, jm.podStoreSynced, jm.jobStoreSynced) {
		return
	}

	for i := 0; i < workers; i++ {
		go wait.Until(jm.worker, time.Second, stopCh)
	}

	<-stopCh
}
202045
func (svc *AsyncService) Run() error {
	for {
		e, open := <-svc.c
		if open {
			// Call event handler
			svc.i.HandleEvent(e)
		} else {
			// Exit runner
			break
		}
	}
	return nil
}
1117417
func (tc *NoExecuteTaintManager) Run(stopCh <-chan struct{}) {
	klog.V(0).Infof("Starting NoExecuteTaintManager")

	for i := 0; i < UpdateWorkerSize; i++ {
		tc.nodeUpdateChannels = append(tc.nodeUpdateChannels, make(chan nodeUpdateItem, NodeUpdateChannelSize))
		tc.podUpdateChannels = append(tc.podUpdateChannels, make(chan podUpdateItem, podUpdateChannelSize))
	}

	// Functions that are responsible for taking work items out of the workqueues and putting them
	// into channels.
	go func(stopCh <-chan struct{}) {
		for {
			item, shutdown := tc.nodeUpdateQueue.Get()
			if shutdown {
				break
			}
			nodeUpdate := item.(nodeUpdateItem)
			hash := hash(nodeUpdate.nodeName, UpdateWorkerSize)
			select {
			case <-stopCh:
				tc.nodeUpdateQueue.Done(item)
				return
			case tc.nodeUpdateChannels[hash] <- nodeUpdate:
				// tc.nodeUpdateQueue.Done is called by the nodeUpdateChannels worker
			}
		}
	}(stopCh)

	go func(stopCh <-chan struct{}) {
		for {
			item, shutdown := tc.podUpdateQueue.Get()
			if shutdown {
				break
			}
			podUpdate := item.(podUpdateItem)
			hash := hash(podUpdate.nodeName, UpdateWorkerSize)
			select {
			case <-stopCh:
				tc.podUpdateQueue.Done(item)
				return
			case tc.podUpdateChannels[hash] <- podUpdate:
				// tc.podUpdateQueue.Done is called by the podUpdateChannels worker
			}
		}
	}(stopCh)

	wg := sync.WaitGroup{}
	wg.Add(UpdateWorkerSize)
	for i := 0; i < UpdateWorkerSize; i++ {
		go tc.worker(i, wg.Done, stopCh)
	}
	wg.Wait()
}
84786
func (worker *copIteratorWorker) run(ctx context.Context) {
	defer worker.wg.Done()
	for task := range worker.taskCh {
		respCh := worker.respChan
		if respCh == nil {
			respCh = task.respChan
		}

		bo := NewBackoffer(ctx, copNextMaxBackoff).WithVars(worker.vars)
		worker.handleTask(bo, task, respCh)
		if bo.totalSleep > 0 {
			metrics.TiKVBackoffHistogram.Observe(float64(bo.totalSleep) / 1000)
		}
		close(task.respChan)
		select {
		case <-worker.finishCh:
			return
		default:
		}
	}
}
268076
func (m *Manager) Run(ctx context.Context) {
	// Save the context for all other jobs usage, and to indicate
	// the manager is running.
	m.mu.Lock()
	if m.ctx != nil {
		m.mu.Unlock()
		panic("Manager is already running")
	}
	m.ctx = ctx
	m.loadAndStartJobsLocked()
	// Signal the successful startup.
	close(m.started)
	m.started = make(chan struct{})
	m.mu.Unlock()

	// Wait for the context to be canceled.
	<-ctx.Done()

	// Clear context and get a copy of the running jobs.
	m.mu.Lock()
	m.ctx = nil
	runningWorkflows := m.workflows
	m.workflows = make(map[string]*runningWorkflow)
	m.mu.Unlock()

	// Abort the running jobs. They won't save their state as
	// m.ctx is nil and they know it means we're shutting down.
	for _, rw := range runningWorkflows {
		rw.cancel()
	}
	for _, rw := range runningWorkflows {
		<-rw.done
	}
}
175775
func (driver *MesosExecutorDriver) Run() (mesosproto.Status, error) {
	driver.lock.Lock()
	defer driver.lock.Unlock()
	return driver.run()
}
444036
func (r *Rabbus) Run(ctx context.Context) error {
	notifyClose := r.NotifyClose(make(chan *amqp.Error))

	for {
		select {
		case m, ok := <-r.emit:
			if !ok {
				return errors.New("unexpected close of emit channel")
			}

			r.produce(m)

		case err := <-notifyClose:
			if err == nil {
				// "… on a graceful close, no error will be sent."
				return nil
			}

			r.handleAmqpClose(err)

			// We have reconnected, so we need a new NotifyClose again.
			notifyClose = r.NotifyClose(make(chan *amqp.Error))

		case <-ctx.Done():
			return ctx.Err()
		}
	}
}
666902
func (worker *Worker) RunJob(jobID string) error {
	qorJob, err := worker.GetJob(jobID)

	if qorJob != nil && err == nil {
		defer func() {
			if r := recover(); r != nil {
				qorJob.AddLog(string(debug.Stack()))
				qorJob.SetProgressText(fmt.Sprint(r))
				qorJob.SetStatus(JobStatusException)
			}
		}()

		if qorJob.GetStatus() != JobStatusNew && qorJob.GetStatus() != JobStatusScheduled {
			return errors.New("invalid job status, current status: " + qorJob.GetStatus())
		}

		if err = qorJob.SetStatus(JobStatusRunning); err == nil {
			if err = qorJob.GetJob().GetQueue().Run(qorJob); err == nil {
				return qorJob.SetStatus(JobStatusDone)
			}

			qorJob.SetProgressText(err.Error())
			qorJob.SetStatus(JobStatusException)
		}
	}

	return err
}
560379
func Run(jobs []Tasker) (err error) {
	// []T does not convert to []Tasker implicitly even is T implements
	// Tasker. We need to iterate on []Tasker making an explicit cast.
	// http://golang.org/doc/faq#convert_slice_of_interface
	prematureEnd := make(chan struct{})
	jobsQueue := make(chan Tasker, workersNumber)
	done := make(chan struct{}, workersNumber)
	var totalDone int
	go populateQueue(jobsQueue, jobs, prematureEnd)
	go parallelizeWorkers(jobsQueue, done)
	// TODO add a case timeout that returns error.
	for {
		select {
		case <-done:
			totalDone++
		case <-prematureEnd:
			err = ErrTasksNotCompleted
		}
		if totalDone == workersNumber {
			// We can assume that jobsQueue is closed and
			// that no goroutine is operating on []Tasker.
			break
		}
	}
	return
}
918699
func (s *Service) Run(c context.Context, f func(context.Context) error) {
	// Log to Stdout using JSON log lines.
	c = teelogger.Use(c, gkelogger.GetFactory(os.Stdout))

	// If a service name isn't specified, default to the base of the current
	// executable.
	if s.Name == "" {
		s.Name = filepath.Base(os.Args[0])
	}
	s.useDatastoreConfig = true // If available, use datastore config.

	rc := 0
	if err := s.runImpl(c, f); err != nil {
		log.WithError(err).Errorf(c, "Application exiting with error.")
		rc = 1
	}
	os.Exit(rc)
}
1187784
func (mf *Consumer) Run(ctx context.Context) error {
	wg := &sync.WaitGroup{}
	jobs := make(chan job)
	mf.startWorkers(ctx, jobs, wg)

	cleanupWG := &sync.WaitGroup{}
	cleanupCtx, cleanupCancel := context.WithCancel(context.Background())
	del := NewBatchDeleter(cleanupCtx, cleanupWG, mf.s, mf.DeleteMessageAccumulatorTimeout, mf.DeleteMessageDrainTimeout)

	messages := make(chan job)
	go mf.receiveMessages(cleanupCtx, cleanupWG, messages, del)
	var done bool
	for {
		// Stop if the context was cancelled
		select {
		case <-ctx.Done():
			close(jobs)
			wg.Wait()
			cleanupCancel()
			cleanupWG.Wait()
			return ctx.Err()
		case msg := <-messages:
			if done {
				break
			}
			select {
			case <-ctx.Done():
				done = true
			case jobs <- msg:
			}
		}
	}
}