// Run starts consumer threads and calls the Collector
// to perform requests. Run blocks while the queue has active requests
485100
func (q *Queue) Run(c *colly.Collector) error {
	wg := &sync.WaitGroup{}
	for i := 0; i < q.Threads; i++ {
		wg.Add(1)
		go func(c *colly.Collector, wg *sync.WaitGroup) {
			defer wg.Done()
			for {
				if q.IsEmpty() {
					if q.activeThreadCount == 0 {
						break
					}
					ch := make(chan bool)
					q.lock.Lock()
					q.threadChans = append(q.threadChans, ch)
					q.lock.Unlock()
					action := <-ch
					if action == stop && q.IsEmpty() {
						break
					}
				}
				q.lock.Lock()
				atomic.AddInt32(&q.activeThreadCount, 1)
				q.lock.Unlock()
				rb, err := q.storage.GetRequest()
				if err != nil || rb == nil {
					q.finish()
					continue
				}
				r, err := c.UnmarshalRequest(rb)
				if err != nil || r == nil {
					q.finish()
					continue
				}
				r.Do()
				q.finish()
			}
		}(c, wg)
	}
	wg.Wait()
	return nil
}
580494
func (a *adapter) Run() error {
	if a.ircEnabled {
		// set up a connection to the IRC gateway
		hal.Logger.Debug("slack - starting IRC connection")
		go a.startIRCConnection()
		hal.Logger.Debug("slack - started IRC connection")
	} else {
		// set up handlers
		hal.Logger.Debug("slack - adding HTTP request handlers")
		hal.Router.HandleFunc("/hal/slack-webhook", a.slackHandler)
		// Someday we won't need this :D
		hal.Router.HandleFunc("/hubot/slack-webhook", a.slackHandler)
		hal.Logger.Debug("slack - added HTTP request handlers")
	}

	hal.Logger.Debugf("slack - channelmode=%v channels=%v", a.channelMode, a.channels)
	return nil
}
452627
func (s *StreamFramer) Run() {
	s.l.Lock()
	defer s.l.Unlock()
	if s.running {
		return
	}

	s.running = true
	go s.run()
}
243054
func (mr *MockImplementorMockRecorder) Run(ctx interface{}) *gomock.Call {
	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "Run", reflect.TypeOf((*MockImplementor)(nil).Run), ctx)
}
166717
func (s preparedGenericAPIServer) Run(stopCh <-chan struct{}) error {
	err := s.NonBlockingRun(stopCh)
	if err != nil {
		return err
	}

	<-stopCh

	err = s.RunPreShutdownHooks()
	if err != nil {
		return err
	}

	// Wait for all requests to finish, which are bounded by the RequestTimeout variable.
	s.HandlerChainWaitGroup.Wait()

	return nil
}
64403
func (c *poolCreateCommand) Run(ctx *cmd.Context) (err error) {
	api, err := c.newAPIFunc()
	if err != nil {
		return err
	}
	defer api.Close()
	return api.CreatePool(c.poolName, c.provider, c.attrs)
}
475336
func (b *broker) run() error {
	listener, err := net.Listen("tcp", *addr)
	if err != nil {
		return err
	}
	defer listener.Close()
	fmt.Printf("Listening on %s\n", listener.Addr())
	for {
		conn, err := listener.Accept()
		if err != nil {
			debugf("Accept error: %v", err)
			continue
		}
		adapter := proton.NewMessagingAdapter(newHandler(&b.queues))
		// We want to accept messages when they are enqueued, not just when they
		// are received, so we turn off auto-accept and prefetch by the adapter.
		adapter.Prefetch = 0
		adapter.AutoAccept = false
		engine, err := proton.NewEngine(conn, adapter)
		if err != nil {
			debugf("Connection error: %v", err)
			continue
		}
		engine.Server() // Enable server-side protocol negotiation.
		debugf("Accepted connection %s", engine)
		go func() { // Start goroutine to run the engine event loop
			engine.Run()
			debugf("Closed %s", engine)
		}()
	}
}
1305136
func (s *Scheduler) Run() {
	s.mu.Lock()
	s.state = stateRunning
	s.mu.Unlock()

	s.startProcessRequestsLoop()
}
1199061
func (h *websocketStreamHandler) run() {
	wg := sync.WaitGroup{}
	wg.Add(len(h.streamPairs))

	for _, pair := range h.streamPairs {
		p := pair
		go func() {
			defer wg.Done()
			h.portForward(p)
		}()
	}

	wg.Wait()
}
1196161
func (h *Interface) run(jobs chan job, ctx context.Context) {

	ListenAndServeIf(h.Name, h, jobs, ctx)
}
400677
func (sm *subscriptionManager) run() error {
	if sm.LoggingEnabled {
		log.Println("SubscriptionManager: Starting run.")
	}
	for {
		// NOTE: we check to see if its time to purge old buffers whenever
		// something happens or a period of inactivity has occurred.
		// An alternative would be to have another goroutine with a
		// select case time.After() but then you'd have concurrency issues
		// with access to the sm.SubEventBuffer and sm.bufferPriorityQueue objs
		// So instead of introducing mutexes we have this uglier manual time check calls
		select {
		case newClient := <-sm.clientSubscriptions:
			sm.handleNewClient(&newClient)
			sm.seeIfTimeToPurgeStaleCategories()
		case disconnected := <-sm.ClientTimeouts:
			sm.handleClientDisconnect(&disconnected)
			sm.seeIfTimeToPurgeStaleCategories()
		case event := <-sm.Events:
			sm.handleNewEvent(&event)
			sm.seeIfTimeToPurgeStaleCategories()
		case <-time.After(time.Duration(5) * time.Second):
			sm.seeIfTimeToPurgeStaleCategories()
		case _ = <-sm.Quit:
			if sm.LoggingEnabled {
				log.Println("SubscriptionManager: received quit signal, stopping.")
			}
			// break out of our infinite loop/select
			return nil
		}
	}
}
968112
func (c *liveStateCache) Run(ctx context.Context) {
	util.RetryUntilSucceed(func() error {
		clusterEventCallback := func(event *db.ClusterEvent) {
			c.lock.Lock()
			defer c.lock.Unlock()
			if cluster, ok := c.clusters[event.Cluster.Server]; ok {
				if event.Type == watch.Deleted {
					cluster.invalidate()
					delete(c.clusters, event.Cluster.Server)
				} else if event.Type == watch.Modified {
					cluster.cluster = event.Cluster
					cluster.invalidate()
				}
			} else if event.Type == watch.Added && isClusterHasApps(c.appInformer.GetStore().List(), event.Cluster) {
				go func() {
					// warm up cache for cluster with apps
					_, _ = c.getSyncedCluster(event.Cluster.Server)
				}()
			}
		}

		return c.db.WatchClusters(ctx, clusterEventCallback)

	}, "watch clusters", ctx, clusterRetryTimeout)

	<-ctx.Done()
}
1068337
func (u *DropsondeUnmarshallerCollection) Run(inputChan <-chan []byte, outputChan chan<- *events.Envelope, waitGroup *sync.WaitGroup) {
	for _, unmarshaller := range u.unmarshallers {
		go func(um *DropsondeUnmarshaller) {
			defer waitGroup.Done()
			um.Run(inputChan, outputChan)
		}(unmarshaller)
	}
}
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
}
299153
func (s *scheduler) run(sections chan uint64, dist chan *request, done chan []byte, quit chan struct{}, wg *sync.WaitGroup) {
	// Create a forwarder channel between requests and responses of the same size as
	// the distribution channel (since that will block the pipeline anyway).
	pend := make(chan uint64, cap(dist))

	// Start the pipeline schedulers to forward between user -> distributor -> user
	wg.Add(2)
	go s.scheduleRequests(sections, dist, pend, quit, wg)
	go s.scheduleDeliveries(pend, done, quit, wg)
}
630403
func (c *controller) Run(stopCh <-chan struct{}) {
	defer utilruntime.HandleCrash()
	go func() {
		<-stopCh
		c.config.Queue.Close()
	}()
	r := NewReflector(
		c.config.ListerWatcher,
		c.config.ObjectType,
		c.config.Queue,
		c.config.FullResyncPeriod,
	)
	r.ShouldResync = c.config.ShouldResync
	r.clock = c.clock

	c.reflectorMutex.Lock()
	c.reflector = r
	c.reflectorMutex.Unlock()

	var wg wait.Group
	defer wg.Wait()

	wg.StartWithChannel(stopCh, r.Run)

	wait.Until(c.processLoop, time.Second, stopCh)
}
1107030
func (eq *eventQueue) run() {
	for {
		block := eq.next()

		if block == nil {
			return // nil block means event queue is closed.
		}

		if err := eq.sink.Write(block...); err != nil {
			logrus.Warnf("eventqueue: error writing events to %v, these events will be lost: %v", eq.sink, err)
		}

		for _, listener := range eq.listeners {
			listener.egress(block...)
		}
	}
}