// run tasks that need to happen once we get a (new) connection
545577
func (a *AutoConnectTransactional) postConnect(newConn NeoConnection) error {
	a.lk.Lock()
	defer a.lk.Unlock()

	a.conn = newConn

	for _, i := range a.indexes {
		if err := a.conn.EnsureIndexes(i); err != nil {
			return fmt.Errorf("failed to apply indexes: %v\n", err)
		}
	}
	a.indexes = nil
	for _, c := range a.constraints {
		if err := a.conn.EnsureConstraints(c); err != nil {
			return fmt.Errorf("failed to apply constraints: %v\n", err)
		}
	}
	a.constraints = nil
	return nil
}
1310353
func (b *Bot) Run() {
	b.RTM = b.Client.NewRTM()
	go b.RTM.ManageConnection()
	for {
		select {
		case msg := <-b.RTM.IncomingEvents:
			ctx := context.Background()
			ctx = AddBotToContext(ctx, b)
			switch ev := msg.Data.(type) {
			case *slack.ConnectedEvent:
				fmt.Printf("Connected: %#v\n", ev.Info.User)
				b.setBotID(ev.Info.User.ID)
			case *slack.MessageEvent:
				// ignore messages from the current user, the bot user
				if b.botUserID == ev.User {
					continue
				}

				ctx = AddMessageToContext(ctx, ev)
				var match RouteMatch
				if matched, ctx := b.Match(ctx, &match); matched {
					match.Handler(ctx)
				}

			case *slack.RTMError:
				fmt.Printf("Error: %s\n", ev.Error())

			case *slack.InvalidAuthEvent:
				fmt.Printf("Invalid credentials")
				break

			default:
				// Ignore other events..
				// fmt.Printf("Unexpected: %v\n", msg.Data)
			}
		}
	}
}
109241
func (r *Orchestrator) Run(ctx context.Context) error {
	defer close(r.doneChan)

	// Watch changes to services and tasks
	queue := r.store.WatchQueue()
	watcher, cancel := queue.Watch()
	defer cancel()

	// Balance existing services and drain initial tasks attached to invalid
	// nodes
	var err error
	r.store.View(func(readTx store.ReadTx) {
		if err = r.initTasks(ctx, readTx); err != nil {
			return
		}

		if err = r.initServices(readTx); err != nil {
			return
		}

		if err = r.initCluster(readTx); err != nil {
			return
		}
	})
	if err != nil {
		return err
	}

	r.tick(ctx)

	for {
		select {
		case event := <-watcher:
			// TODO(stevvooe): Use ctx to limit running time of operation.
			r.handleTaskEvent(ctx, event)
			r.handleServiceEvent(ctx, event)
			switch v := event.(type) {
			case state.EventCommit:
				r.tick(ctx)
			case api.EventUpdateCluster:
				r.cluster = v.Cluster
			}
		case <-r.stopChan:
			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()
}
1022339
func (c *Controller) Run(workers int, stopCh <-chan struct{}) {
	defer utilruntime.HandleCrash()
	defer c.queue.ShutDown()

	klog.Infof("Starting controller")

	if !cache.WaitForCacheSync(stopCh, c.syncs...) {
		utilruntime.HandleError(fmt.Errorf("timed out waiting for caches to sync"))
		return
	}

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

	<-stopCh
	klog.Infof("Shutting down controller")
}
669969
func (s *statsClient) Run(ctx context.Context) error {
	if err := s.rpchdl.NewRPCClient(statsContextID, s.statsChannel, s.secret); err != nil {
		zap.L().Error("Stats RPC client cannot connect", zap.Error(err))
		return err
	}

	go s.sendStats(ctx)

	return nil
}
1320543
func (h *httpStreamHandler) run() {
	klog.V(5).Infof("(conn=%p) waiting for port forward streams", h.conn)
Loop:
	for {
		select {
		case <-h.conn.CloseChan():
			klog.V(5).Infof("(conn=%p) upgraded connection closed", h.conn)
			break Loop
		case stream := <-h.streamChan:
			requestID := h.requestID(stream)
			streamType := stream.Headers().Get(api.StreamType)
			klog.V(5).Infof("(conn=%p, request=%s) received new stream of type %s", h.conn, requestID, streamType)

			p, created := h.getStreamPair(requestID)
			if created {
				go h.monitorStreamPair(p, time.After(h.streamCreationTimeout))
			}
			if complete, err := p.add(stream); err != nil {
				msg := fmt.Sprintf("error processing stream for request %s: %v", requestID, err)
				utilruntime.HandleError(errors.New(msg))
				p.printError(msg)
			} else if complete {
				go h.portForward(p)
			}
		}
	}
}
1346561
func (ar *allocRunner) Run() {
	// Close the wait channel on return
	defer close(ar.waitCh)

	// Start the task state update handler
	go ar.handleTaskStateUpdates()

	// Start the alloc update handler
	go ar.handleAllocUpdates()

	// If task update chan has been closed, that means we've been shutdown.
	select {
	case <-ar.taskStateUpdateHandlerCh:
		return
	default:
	}

	// Run the prestart hooks if non-terminal
	if ar.shouldRun() {
		if err := ar.prerun(); err != nil {
			ar.logger.Error("prerun failed", "error", err)
			goto POST
		}
	}

	// Run the runners (blocks until they exit)
	ar.runTasks()

POST:
	// Run the postrun hooks
	if err := ar.postrun(); err != nil {
		ar.logger.Error("postrun failed", "error", err)
	}

}
1217519
func (hub *Hub) run() {
	if hub.isRunning != true { // Should be safe, because only called from NewRouter and therefore a single thread.
		hub.isRunning = true
		go func() {
			for {
				select {
				// Register new connection
				case conn := <-hub.register:
					hub.connections[conn] = true
				// Unregister dropped connection
				case conn := <-hub.unregister:
					if _, ok := hub.connections[conn]; ok {
						hub.remove(conn)
					}
				// Broadcast
				case message := <-hub.broadcast:
					for conn := range hub.connections {
						select {
						case conn.send <- message:
						default:
							hub.remove(conn)
						}
					}
				}
			}
		}()
	}
}
203261
func (dht *DHT) Run() {
	dht.init()
	dht.listen()
	dht.join()

	dht.Ready = true

	var pkt packet
	tick := time.Tick(dht.CheckKBucketPeriod)

	for {
		select {
		case pkt = <-dht.packets:
			handle(dht, pkt)
		case <-tick:
			if dht.routingTable.Len() == 0 {
				dht.join()
			} else if dht.transactionManager.len() == 0 {
				go dht.routingTable.Fresh()
			}
		}
	}
}
873463
func (self *RedisBackend) run(cmd string, args ...interface{}) (interface{}, error) {
	if conn := self.pool.Get(); conn != nil {
		defer conn.Close()

		// debug := strings.Join(sliceutil.Stringify(args), ` `)
		// querylog.Debugf("[%v] %v %v", self, cmd, debug)
		return redis.DoWithTimeout(conn, self.cmdTimeout, cmd, args...)
	} else {
		return nil, fmt.Errorf("Failed to borrow Redis connection")
	}
}
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
}
644668
func (s *ScheduledImageStreamController) Run(stopCh <-chan struct{}) {
	defer utilruntime.HandleCrash()

	klog.Infof("Starting scheduled import controller")

	// Wait for the stream store to sync before starting any work in this controller.
	if !cache.WaitForCacheSync(stopCh, s.listerSynced) {
		utilruntime.HandleError(fmt.Errorf("timed out waiting for caches to sync"))
		return
	}

	go s.scheduler.RunUntil(stopCh)

	metrics.InitializeImportCollector(true, s.importCounter.Collect)

	<-stopCh
	klog.Infof("Shutting down image stream controller")
}
365799
func (p *ParallelRunner) Run() error {
	// default value is 0. The task will not run in this case.
	var parallelNum int
	switch p.concurrencyLevel {
	case Sequential:
		parallelNum = 1
	case Parallel:
		parallelNum = len(p.tasks)
	default:
		log.Fatalf("BUG: Invalid concurrency level: %v", p.concurrencyLevel)
	}
	// sem is a channel used to control the level of concurrency.
	sem := make(chan bool, parallelNum)
	wg := sync.WaitGroup{}
	for i, task := range p.tasks {
		if isTaskSucceeded(task) {
			continue
		}

		sem <- true
		if p.enableApprovals && !isTaskRunning(task) {
			p.waitForApproval(i)
		}
		select {
		case <-p.ctx.Done():
			// Break this run and return early. Do not try to execute any subsequent tasks.
			log.Infof("Workflow is cancelled, remaining tasks will be aborted")
			return nil
		default:
			wg.Add(1)
			go func(t *workflowpb.Task) {
				defer wg.Done()
				p.setUIMessage(fmt.Sprintf("Launch task: %v.", t.Id))
				defer func() { <-sem }()
				p.executeTask(t)
			}(task)
		}
	}
	wg.Wait()

	if p.enableApprovals {
		p.clearPhaseActions()
	}
	// TODO(yipeiw): collect error message from tasks.Error instead,
	// s.t. if the task is retried, we can update the error
	return nil
}
63578
func (sc *syncClient) run() error {
	for {
		select {
		case <-sc.stopped:
			<-sc.start
			log.Infof("Starting sync client...")
		default:
			sc.runOnce()
			select {
			case <-time.After(5 * time.Second):
			case <-sc.quit:
				sc.stopped <- true
			}
		}
	}
}
803046
func main() {
	// DEBUG
	runtime.GOMAXPROCS(2)

	log.SetPrefix("[go-redis|bench] ")
	flag.Parse()

	fmt.Printf("\n\n== Bench synchclient == %d goroutines 1 AsyncClient  -- %d opts each --- \n\n", *workers, *opcnt)

	for _, task := range tasks {
		// REVU: creates a new client for each task run
		// wondering if using the same conn throughout will be more realistic?
		// regardless flushdb per task is OK
		benchTask(task, *opcnt, *workers, true)
	}

}
300413
func (s *Streamer) run() {
	go func() {
		for {
			select {
			case cl := <-s.connecting:
				s.clients[cl] = true

			case cl := <-s.disconnecting:
				delete(s.clients, cl)

			case event := <-s.event:
				for cl := range s.clients {
					// TODO: non-blocking broadcast
					//select {
					//case cl <- event: // Try to send event to client
					//default:
					//	fmt.Println("Channel full. Discarding value")
					//}
					cl <- event
				}
			}
		}
	}()
}