// Run starts the event loop that pushes Prometheus metrics to Graphite at the
// configured interval.
345805
func (b *Bridge) Run(ctx context.Context) {
	ticker := time.NewTicker(b.interval)
	defer ticker.Stop()
	for {
		select {
		case <-ticker.C:
			if err := b.Push(); err != nil && b.logger != nil {
				b.logger.Println("error pushing to Graphite:", err)
			}
		case <-ctx.Done():
			return
		}
	}
}
713189
func (c *AggregationController) Run(stopCh <-chan struct{}) {
	defer utilruntime.HandleCrash()
	defer c.queue.ShutDown()

	klog.Infof("Starting OpenAPI AggregationController")
	defer klog.Infof("Shutting down OpenAPI AggregationController")

	go wait.Until(c.runWorker, time.Second, stopCh)

	<-stopCh
}
190744
func (c *Consumer) Run(projection *sqlsol.Projection, abiSpec *abi.AbiSpec, stream bool) error {
	var err error

	c.Log.Info("msg", "Connecting to Burrow gRPC server")

	c.GRPCConnection, err = grpc.Dial(c.Config.GRPCAddr, grpc.WithInsecure())
	if err != nil {
		return errors.Wrapf(err, "Error connecting to Burrow gRPC server at %s", c.Config.GRPCAddr)
	}
	defer c.GRPCConnection.Close()
	defer close(c.EventsChannel)

	// get the chain ID to compare with the one stored in the db
	qCli := rpcquery.NewQueryClient(c.GRPCConnection)
	chainStatus, err := qCli.Status(context.Background(), &rpcquery.StatusParam{})
	if err != nil {
		return errors.Wrapf(err, "Error getting chain status")
	}

	if len(projection.EventSpec) == 0 {
		c.Log.Info("msg", "No events specifications found")
		return nil
	}

	c.Log.Info("msg", "Connecting to SQL database")

	connection := types.SQLConnection{
		DBAdapter:     c.Config.DBAdapter,
		DBURL:         c.Config.DBURL,
		DBSchema:      c.Config.DBSchema,
		Log:           c.Log,
		ChainID:       chainStatus.ChainID,
		BurrowVersion: chainStatus.BurrowVersion,
	}

	c.DB, err = sqldb.NewSQLDB(connection)
	if err != nil {
		return fmt.Errorf("error connecting to SQL database: %v", err)
	}
	defer c.DB.Close()

	c.Log.Info("msg", "Synchronizing config and database projection structures")

	err = c.DB.SynchronizeDB(projection.Tables)
	if err != nil {
		return errors.Wrap(err, "Error trying to synchronize database")
	}

	// doneCh is used for sending a "done" signal from each goroutine to the main thread
	// eventCh is used for sending received events to the main thread to be stored in the db
	doneCh := make(chan struct{})
	errCh := make(chan error, 1)
	eventCh := make(chan types.EventData)

	go func() {
		defer func() {
			close(doneCh)
		}()
		go c.announceEvery(doneCh)

		c.Log.Info("msg", "Getting last processed block number from SQL log table")

		// NOTE [Silas]: I am preserving the comment below that dates from the early days of Vent. I have looked at the
		// bosmarmot git history and I cannot see why the original author thought that it was the case that there was
		// no way of knowing if the last block of events was committed since the block and its associated log is
		// committed atomically in a transaction and this is a core part of he design of Vent - in order that it does not
		// repeat

		// [ORIGINAL COMMENT]
		// right now there is no way to know if the last block of events was completely read
		// so we have to begin processing from the last block number stored in database
		// and update event data if already present
		fromBlock, err := c.DB.GetLastBlockHeight()
		if err != nil {
			errCh <- errors.Wrapf(err, "Error trying to get last processed block number from SQL log table")
			return
		}

		startingBlock := fromBlock
		// Start the block after the last one successfully committed - apart from if this is the first block
		// We include block 0 because it is where we currently place dump/restored transactions
		if startingBlock > 0 {
			startingBlock++
		}

		// setup block range to get needed blocks server side
		cli := rpcevents.NewExecutionEventsClient(c.GRPCConnection)
		var end *rpcevents.Bound
		if stream {
			end = rpcevents.StreamBound()
		} else {
			end = rpcevents.LatestBound()
		}

		request := &rpcevents.BlocksRequest{
			BlockRange: rpcevents.NewBlockRange(rpcevents.AbsoluteBound(startingBlock), end),
		}

		// gets blocks in given range based on last processed block taken from database
		stream, err := cli.Stream(context.Background(), request)
		if err != nil {
			errCh <- errors.Wrapf(err, "Error connecting to block stream")
			return
		}

		// get blocks

		c.Log.Debug("msg", "Waiting for blocks...")

		err = rpcevents.ConsumeBlockExecutions(stream, c.makeBlockConsumer(projection, abiSpec, eventCh))

		if err != nil {
			if err == io.EOF {
				c.Log.Info("msg", "EOF stream received...")
			} else {
				if c.Closing {
					c.Log.Debug("msg", "GRPC connection closed")
				} else {
					errCh <- errors.Wrapf(err, "Error receiving blocks")
					return
				}
			}
		}
	}()

	for {
		select {
		// Process block events
		case blk := <-eventCh:
			err := c.commitBlock(projection, blk)
			if err != nil {
				c.Log.Info("msg", "error committing block", "err", err)
				return err
			}

		// Await completion
		case <-doneCh:
			select {

			// Select possible error
			case err := <-errCh:
				c.Log.Info("msg", "finished with error", "err", err)
				return err

			// Or fallback to success
			default:
				c.Log.Info("msg", "finished successfully")
				return nil
			}
		}
	}
}
577572
func (s *StepperDriver) Run() (err error) {
	//halt if already moving
	if s.moving == true {
		s.Halt()
	}

	s.mutex.Lock()
	s.moving = true
	s.mutex.Unlock()

	go func() {
		for {
			if s.moving == false {
				break
			}
			s.step()
		}
	}()

	return
}
289637
func run(ctx context.Context, options GlobalOptions) {
	logrus.Info("Starting honeytail")

	stats := newResponseStats()

	sigs := make(chan os.Signal, 1)
	ctx, cancel := context.WithCancel(ctx)
	signal.Notify(sigs, syscall.SIGINT, syscall.SIGTERM)

	// spin up our transmission to send events to Honeycomb
	libhConfig := libhoney.Config{
		WriteKey:             options.Reqs.WriteKey,
		Dataset:              options.Reqs.Dataset,
		APIHost:              options.APIHost,
		MaxConcurrentBatches: options.NumSenders,
		SendFrequency:        time.Duration(options.BatchFrequencyMs) * time.Millisecond,
		MaxBatchSize:         options.BatchSize,
		// block on send should be true so if we can't send fast enough, we slow
		// down reading the log rather than drop lines.
		BlockOnSend: true,
		// block on response is true so that if we hit rate limiting we make sure
		// to re-enqueue all dropped events
		BlockOnResponse: true,

		// limit pending work capacity so that we get backpressure from libhoney
		// and block instead of sleeping inside sendToLibHoney.
		PendingWorkCapacity: 20 * options.NumSenders,
	}
	if options.DebugOut {
		libhConfig.Output = &libhoney.WriterOutput{}
	}
	if err := libhoney.Init(libhConfig); err != nil {
		logrus.WithFields(logrus.Fields{"err": err}).Fatal(
			"Error occured while spinning up Transimission")
	}

	// compile the prefix regex once for use on all channels
	var prefixRegex *parsers.ExtRegexp
	if options.PrefixRegex == "" {
		prefixRegex = nil
	} else {
		prefixRegex = &parsers.ExtRegexp{regexp.MustCompile(options.PrefixRegex)}
	}

	// get our lines channel from which to read log lines
	var linesChans []chan string
	var err error
	tc := tail.Config{
		Paths:       options.Reqs.LogFiles,
		FilterPaths: options.FilterFiles,
		Type:        tail.RotateStyleSyslog,
		Options:     options.Tail,
	}
	if options.TailSample {
		linesChans, err = tail.GetSampledEntries(ctx, tc, options.SampleRate)
	} else {
		linesChans, err = tail.GetEntries(ctx, tc)
	}
	if err != nil {
		logrus.WithFields(logrus.Fields{"err": err}).Fatal(
			"Error occurred while trying to tail logfile")
	}

	// set up our signal handler and support canceling
	go func() {
		sig := <-sigs
		fmt.Fprintf(os.Stderr, "Aborting! Caught signal \"%s\"\n", sig)
		fmt.Fprintf(os.Stderr, "Cleaning up...\n")
		cancel()
		// and if they insist, catch a second CTRL-C or timeout on 10sec
		select {
		case <-sigs:
			fmt.Fprintf(os.Stderr, "Caught second signal... Aborting.\n")
			os.Exit(1)
		case <-time.After(10 * time.Second):
			fmt.Fprintf(os.Stderr, "Taking too long... Aborting.\n")
			os.Exit(1)
		}
	}()

	// for each channel we got back from tail.GetEntries, spin up a parser.
	parsersWG := sync.WaitGroup{}
	responsesWG := sync.WaitGroup{}
	for _, lines := range linesChans {
		// get our parser
		parser, opts := getParserAndOptions(options)
		if parser == nil {
			logrus.WithFields(logrus.Fields{"parser": options.Reqs.ParserName}).Fatal(
				"Parser not found. Use --list to show valid parsers")
		}

		// and initialize it
		if err := parser.Init(opts); err != nil {
			logrus.Fatalf(
				"Error initializing %s parser module: %v", options.Reqs.ParserName, err)
		}

		// create a channel for sending events into libhoney
		toBeSent := make(chan event.Event, options.NumSenders)
		doneSending := make(chan bool)

		// two channels to handle backing off when rate limited and resending failed
		// send attempts that are recoverable
		toBeResent := make(chan event.Event, 2*options.NumSenders)
		// time in milliseconds to delay the send
		delaySending := make(chan int, 2*options.NumSenders)

		// apply any filters to the events before they get sent
		modifiedToBeSent := modifyEventContents(toBeSent, options)

		realToBeSent := make(chan event.Event, 10*options.NumSenders)
		go func() {
			wg := sync.WaitGroup{}
			for i := uint(0); i < options.NumSenders; i++ {
				wg.Add(1)
				go func() {
					for ev := range modifiedToBeSent {
						realToBeSent <- ev
					}
					wg.Done()
				}()
			}
			wg.Wait()
			close(realToBeSent)
		}()

		// start up the sender. all sources are either sampled when tailing or in-
		// parser, so always tell libhoney events are pre-sampled
		go sendToLibhoney(ctx, realToBeSent, toBeResent, delaySending, doneSending)

		// start a goroutine that reads from responses and logs.
		responses := libhoney.Responses()
		responsesWG.Add(1)
		go func() {
			handleResponses(responses, stats, toBeResent, delaySending, options)
			responsesWG.Done()
		}()

		parsersWG.Add(1)
		go func(plines chan string) {
			// ProcessLines won't return until lines is closed
			parser.ProcessLines(plines, toBeSent, prefixRegex)
			// trigger the sending goroutine to finish up
			close(toBeSent)
			// wait for all the events in toBeSent to be handed to libhoney
			<-doneSending
			parsersWG.Done()
		}(lines)
	}
	parsersWG.Wait()
	// tell libhoney to finish up sending events
	libhoney.Close()
	// print out what we've done one last time
	responsesWG.Wait()
	stats.log()
	stats.logFinal()

	// Nothing bad happened, yay
	logrus.Info("Honeytail is all done, goodbye!")
}
918956
func (t *ThrottledExecutor) Run(ctx context.Context, job func()) {
	t.wg.Add(1)
	go func() {
		defer t.wg.Done()
		select {
		case t.limiter <- struct{}{}:
			defer func() {
				<-t.limiter
			}()
			job()
		case <-ctx.Done():
			return
		}
	}()
}
224173
func (mod *TwilioBee) Run(eventChan chan bees.Event) {
	mod.client = twilio.NewClient(mod.accountsid, mod.authtoken)
}
1107838
func (e *BackgroundTask) Run(m libkb.MetaContext) (err error) {
	defer m.Trace(e.Name(), func() error { return err })()

	// use a new background context with a saved cancel function
	var cancel func()
	m, cancel = m.BackgroundWithCancel()

	e.Lock()
	defer e.Unlock()

	e.shutdownFunc = cancel
	if e.shutdown {
		// Shutdown before started
		cancel()
		e.meta("early-shutdown")
		return nil
	}

	// start the loop and return
	go func() {
		err := e.loop(m)
		if err != nil {
			e.log(m, "loop error: %s", err)
		}
		cancel()
		e.meta("loop-exit")
	}()

	return nil
}
81814
func (c *SetMeterStatusCommand) Run(ctx *cmd.Context) error {
	client, err := newClient(c.ModelCommandBase)
	if err != nil {
		return errors.Trace(err)
	}
	defer client.Close()
	err = client.SetMeterStatus(c.Tag.String(), c.Status, c.StatusInfo)
	if err != nil {
		return errors.Trace(err)
	}
	return nil
}
461957
func (a *AllocGarbageCollector) Run() {
	ticker := time.NewTicker(a.config.Interval)
	for {
		select {
		case <-a.triggerCh:
		case <-ticker.C:
		case <-a.shutdownCh:
			ticker.Stop()
			return
		}

		if err := a.keepUsageBelowThreshold(); err != nil {
			a.logger.Error("error garbage collecting allocations", "error", err)
		}
	}
}
71451
func (e *Epoller) Run(ctx context.Context) {
	var events [128]unix.EpollEvent
	for {
		select {
		case <-ctx.Done():
			e.Close()
			return
		default:
			n, err := unix.EpollWait(e.fd, events[:], -1)
			if err != nil {
				if err == unix.EINTR {
					continue
				}
				logrus.WithError(err).Error("cgroups: epoll wait")
			}
			for i := 0; i < n; i++ {
				e.process(ctx, uintptr(events[i].Fd))
			}
		}
	}
}
454855
func (p *PeriodicDispatch) run(ctx context.Context, updateCh <-chan struct{}) {
	var launchCh <-chan time.Time
	for p.shouldRun() {
		job, launch := p.nextLaunch()
		if launch.IsZero() {
			launchCh = nil
		} else {
			launchDur := launch.Sub(time.Now().In(job.Periodic.GetLocation()))
			launchCh = time.After(launchDur)
			p.logger.Debug("scheduled periodic job launch", "launch_delay", launchDur, "job", job.NamespacedID())
		}

		select {
		case <-ctx.Done():
			return
		case <-updateCh:
			continue
		case <-launchCh:
			p.dispatch(job, launch)
		}
	}
}
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:
		}
	}
}
425178
func (c *MetricsCommand) Run(ctx *cmd.Context) error {
	client, err := newClient(c.ModelCommandBase)
	if err != nil {
		return errors.Trace(err)
	}
	var metrics []params.MetricResult
	if c.All {
		metrics, err = client.GetMetrics()
	} else {
		metrics, err = client.GetMetrics(c.Tags...)
	}
	if err != nil {
		return errors.Trace(err)
	}
	defer client.Close()
	if len(metrics) == 0 {
		return nil
	}
	results := make([]metric, len(metrics))
	for i, m := range metrics {
		results[i] = metric{
			Unit:      m.Unit,
			Timestamp: m.Time,
			Metric:    m.Key,
			Value:     m.Value,
			Labels:    m.Labels,
		}
	}
	sortedResults := metricSlice(results)
	sort.Sort(sortedResults)

	return errors.Trace(c.out.Write(ctx, results))
}
470802
func (e *PaperKeyGen) Run(m libkb.MetaContext) error {
	if !e.arg.SkipPush {
		err := e.syncPUK(m)
		if err != nil {
			return err
		}
	}

	// make the passphrase stream
	key, err := scrypt.Key(e.arg.Passphrase.Bytes(), nil,
		libkb.PaperKeyScryptCost, libkb.PaperKeyScryptR, libkb.PaperKeyScryptP, libkb.PaperKeyScryptKeylen)
	if err != nil {
		return err
	}

	ppStream := libkb.NewPassphraseStream(key)

	// make keys for the paper device
	if err := e.makeSigKey(ppStream.EdDSASeed()); err != nil {
		return err
	}
	if err := e.makeEncKey(ppStream.DHSeed()); err != nil {
		return err
	}

	// push everything to the server
	if err := e.push(m); err != nil {
		return err
	}

	// no need to notify if key wasn't pushed to server
	// (e.g. in the case of using this engine to verify a key)
	if e.arg.SkipPush {
		return nil
	}

	e.G().KeyfamilyChanged(m.Ctx(), e.getUID())

	return nil
}
96539
func (d *DockerMonitor) Run(ctx context.Context) error {

	if err := d.config.IsComplete(); err != nil {
		return fmt.Errorf("docker: %s", err)
	}

	err := d.waitForDockerDaemon(ctx)
	if err != nil {
		zap.L().Error("Docker daemon is not running - skipping container processing", zap.Error(err))
		return nil
	}

	if d.syncAtStart && d.config.Policy != nil {

		options := types.ContainerListOptions{All: true}
		containers, err := d.dockerClient.ContainerList(ctx, options)
		if err != nil {
			return fmt.Errorf("unable to get container list: %s", err)
		}

		// Starting the eventListener and wait to hear on channel for it to be ready.
		// Need to start before the resync process so that we don't loose any events.
		// They will be buffered. We don't want to start the listener before
		// getting the list from docker though, to avoid duplicates.
		listenerReady := make(chan struct{})
		go d.eventListener(ctx, listenerReady)
		<-listenerReady

		zap.L().Debug("Syncing all existing containers")
		// Syncing all Existing containers depending on MonitorSetting
		if err := d.resyncContainers(ctx, containers); err != nil {
			zap.L().Error("Unable to sync existing containers", zap.Error(err))
		}
	} else {
		// Starting the eventListener and wait to hear on channel for it to be ready.
		// We are not doing resync. We just start the listener.
		listenerReady := make(chan struct{})
		go d.eventListener(ctx, listenerReady)
		<-listenerReady
	}

	// Start processing the events
	go d.eventProcessors(ctx)

	return nil
}
314815
func (mod *HTTPBee) Run(cin chan bees.Event) {
	mod.eventChan = cin

	select {
	case <-mod.SigChan:
		return
	}
}