// Run connects to a grpc service and subscribes to log events,
// then gets tables structures, maps them & parse event data.
// Store data in SQL event tables, it runs forever
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
			}
		}
	}
}
195650
func (d *Dispatcher) Run(ctx context.Context) error {
	ctx = log.WithModule(ctx, "dispatcher")
	log.G(ctx).Info("dispatcher starting")

	d.taskUpdatesLock.Lock()
	d.taskUpdates = make(map[string]*api.TaskStatus)
	d.taskUpdatesLock.Unlock()

	d.nodeUpdatesLock.Lock()
	d.nodeUpdates = make(map[string]nodeUpdate)
	d.nodeUpdatesLock.Unlock()

	d.mu.Lock()
	if d.isRunning() {
		d.mu.Unlock()
		return errors.New("dispatcher is already running")
	}
	if err := d.markNodesUnknown(ctx); err != nil {
		log.G(ctx).Errorf(`failed to move all nodes to "unknown" state: %v`, err)
	}
	configWatcher, cancel, err := store.ViewAndWatch(
		d.store,
		func(readTx store.ReadTx) error {
			clusters, err := store.FindClusters(readTx, store.ByName(store.DefaultClusterName))
			if err != nil {
				return err
			}
			if err == nil && len(clusters) == 1 {
				heartbeatPeriod, err := gogotypes.DurationFromProto(clusters[0].Spec.Dispatcher.HeartbeatPeriod)
				if err == nil && heartbeatPeriod > 0 {
					d.config.HeartbeatPeriod = heartbeatPeriod
				}
				if clusters[0].NetworkBootstrapKeys != nil {
					d.networkBootstrapKeys = clusters[0].NetworkBootstrapKeys
				}
				d.lastSeenRootCert = clusters[0].RootCA.CACert
			}
			return nil
		},
		api.EventUpdateCluster{},
	)
	if err != nil {
		d.mu.Unlock()
		return err
	}
	// set queue here to guarantee that Close will close it
	d.clusterUpdateQueue = watch.NewQueue()

	peerWatcher, peerCancel := d.cluster.SubscribePeers()
	defer peerCancel()
	d.lastSeenManagers = getWeightedPeers(d.cluster)

	defer cancel()
	d.ctx, d.cancel = context.WithCancel(ctx)
	ctx = d.ctx
	d.wg.Add(1)
	defer d.wg.Done()
	d.mu.Unlock()

	publishManagers := func(peers []*api.Peer) {
		var mgrs []*api.WeightedPeer
		for _, p := range peers {
			mgrs = append(mgrs, &api.WeightedPeer{
				Peer:   p,
				Weight: remotes.DefaultObservationWeight,
			})
		}
		d.mu.Lock()
		d.lastSeenManagers = mgrs
		d.mu.Unlock()
		d.clusterUpdateQueue.Publish(clusterUpdate{managerUpdate: &mgrs})
	}

	batchTimer := time.NewTimer(maxBatchInterval)
	defer batchTimer.Stop()

	for {
		select {
		case ev := <-peerWatcher:
			publishManagers(ev.([]*api.Peer))
		case <-d.processUpdatesTrigger:
			d.processUpdates(ctx)
			batchTimer.Stop()
			// drain the timer, if it has already expired
			select {
			case <-batchTimer.C:
			default:
			}
			batchTimer.Reset(maxBatchInterval)
		case <-batchTimer.C:
			d.processUpdates(ctx)
			// batch timer has already expired, so no need to drain
			batchTimer.Reset(maxBatchInterval)
		case v := <-configWatcher:
			cluster := v.(api.EventUpdateCluster)
			d.mu.Lock()
			if cluster.Cluster.Spec.Dispatcher.HeartbeatPeriod != nil {
				// ignore error, since Spec has passed validation before
				heartbeatPeriod, _ := gogotypes.DurationFromProto(cluster.Cluster.Spec.Dispatcher.HeartbeatPeriod)
				if heartbeatPeriod != d.config.HeartbeatPeriod {
					// only call d.nodes.updatePeriod when heartbeatPeriod changes
					d.config.HeartbeatPeriod = heartbeatPeriod
					d.nodes.updatePeriod(d.config.HeartbeatPeriod, d.config.HeartbeatEpsilon, d.config.GracePeriodMultiplier)
				}
			}
			d.lastSeenRootCert = cluster.Cluster.RootCA.CACert
			d.networkBootstrapKeys = cluster.Cluster.NetworkBootstrapKeys
			d.mu.Unlock()
			d.clusterUpdateQueue.Publish(clusterUpdate{
				bootstrapKeyUpdate: &cluster.Cluster.NetworkBootstrapKeys,
				rootCAUpdate:       &cluster.Cluster.RootCA.CACert,
			})
		case <-ctx.Done():
			return nil
		}
	}
}
791967
func (m *Main) Run() error {
	m.Logger.Info("running")

	// Read or generate the instance identifier from file.
	id, err := m.readID(filepath.Join(m.DataDir, "instance_id"))
	if err != nil {
		return err
	}
	m.Process.ID = id
	m.Process.DataDir = m.DataDir
	m.Process.Logger = m.Logger.New("component", "process", "id", id)

	// Start process.
	m.Logger.Info("starting process", "id", id)
	if err := m.Process.Start(); err != nil {
		m.Logger.Error("error starting process", "err", err)
		return err
	}

	// Add service to discoverd registry.
	m.Logger.Info("adding service", "name", m.ServiceName)
	if err = m.DiscoverdClient.AddService(m.ServiceName, nil); err != nil && !httphelper.IsObjectExistsError(err) {
		m.Logger.Error("error adding discoverd service", "err", err)
		return err
	}
	inst := &discoverd.Instance{
		Addr: ":" + m.Process.Port,
		Meta: map[string]string{"REDIS_ID": id},
	}

	// Register instance and retain heartbeater.
	m.Logger.Info("registering instance", "addr", inst.Addr, "meta", inst.Meta)
	hb, err := m.DiscoverdClient.RegisterInstance(m.ServiceName, inst)
	if err != nil {
		m.Logger.Error("error registering discoverd instance", "err", err)
		return err
	}
	m.hb = hb
	shutdown.BeforeExit(func() { hb.Close() })

	m.Logger.Info("opening port", "addr", m.Addr)

	// Open HTTP port.
	ln, err := net.Listen("tcp", m.Addr)
	if err != nil {
		m.Logger.Error("error opening port", "err", err)
		return err
	}
	m.ln = keepalive.Listener(ln)

	// Initialize and server handler.
	m.Logger.Info("serving http api")
	h := redis.NewHandler()
	h.Process = m.Process
	h.Heartbeater = m.hb
	h.Logger = m.Logger.New("component", "http")
	go func() { http.Serve(ln, h) }()

	return nil
}
981854
func Run(t crossdock.T) {
	encoding := t.Param(params.Encoding)
	f, ok := encodingToRunFunc[encoding]
	if !ok {
		crossdock.Fatals(t).Fail("unknown encoding", "%v", encoding)
	}
	f(t, "grpc")
}
426377
func (g *Goon) Run(q *datastore.Query) *Iterator {
	return &Iterator{
		g: g,
		i: q.Run(g.Context),
	}
}
1237000
func (a *ArgoCDServer) Run(ctx context.Context, port int) {
	grpcS := a.newGRPCServer()
	grpcWebS := grpcweb.WrapServer(grpcS)
	var httpS *http.Server
	var httpsS *http.Server
	if a.useTLS() {
		httpS = newRedirectServer(port)
		httpsS = a.newHTTPServer(ctx, port, grpcWebS)
	} else {
		httpS = a.newHTTPServer(ctx, port, grpcWebS)
	}
	metricsServ := newAPIServerMetricsServer()

	// Start listener
	var conn net.Listener
	var realErr error
	_ = wait.ExponentialBackoff(backoff, func() (bool, error) {
		conn, realErr = net.Listen("tcp", fmt.Sprintf(":%d", port))
		if realErr != nil {
			a.log.Warnf("failed listen: %v", realErr)
			return false, nil
		}
		return true, nil
	})
	errors.CheckError(realErr)

	// Cmux is used to support servicing gRPC and HTTP1.1+JSON on the same port
	tcpm := cmux.New(conn)
	var tlsm cmux.CMux
	var grpcL net.Listener
	var httpL net.Listener
	var httpsL net.Listener
	if !a.useTLS() {
		httpL = tcpm.Match(cmux.HTTP1Fast())
		grpcL = tcpm.Match(cmux.HTTP2HeaderField("content-type", "application/grpc"))
	} else {
		// We first match on HTTP 1.1 methods.
		httpL = tcpm.Match(cmux.HTTP1Fast())

		// If not matched, we assume that its TLS.
		tlsl := tcpm.Match(cmux.Any())
		tlsConfig := tls.Config{
			Certificates: []tls.Certificate{*a.settings.Certificate},
		}
		if a.TLSConfigCustomizer != nil {
			a.TLSConfigCustomizer(&tlsConfig)
		}
		tlsl = tls.NewListener(tlsl, &tlsConfig)

		// Now, we build another mux recursively to match HTTPS and gRPC.
		tlsm = cmux.New(tlsl)
		httpsL = tlsm.Match(cmux.HTTP1Fast())
		grpcL = tlsm.Match(cmux.Any())
	}

	// Start the muxed listeners for our servers
	log.Infof("argocd %s serving on port %d (url: %s, tls: %v, namespace: %s, sso: %v)",
		argocd.GetVersion(), port, a.settings.URL, a.useTLS(), a.Namespace, a.settings.IsSSOConfigured())

	go a.projInformer.Run(ctx.Done())
	go func() { a.checkServeErr("grpcS", grpcS.Serve(grpcL)) }()
	go func() { a.checkServeErr("httpS", httpS.Serve(httpL)) }()
	if a.useTLS() {
		go func() { a.checkServeErr("httpsS", httpsS.Serve(httpsL)) }()
		go func() { a.checkServeErr("tlsm", tlsm.Serve()) }()
	}
	go a.watchSettings(ctx)
	go a.rbacPolicyLoader(ctx)
	go func() { a.checkServeErr("tcpm", tcpm.Serve()) }()
	go func() { a.checkServeErr("metrics", metricsServ.ListenAndServe()) }()
	if !cache.WaitForCacheSync(ctx.Done(), a.projInformer.HasSynced) {
		log.Fatal("Timed out waiting for project cache to sync")
	}

	a.stopCh = make(chan struct{})
	<-a.stopCh
	errors.CheckError(conn.Close())
}
688470
func run() {
	log.Info("Logging queries to syslog")
	defer writer.Close()

	// ch will only be non-nil in a unit test context, when a mock has been populated
	if ch == nil {
		ch = tabletenv.StatsLogger.Subscribe("gwslog")
		defer tabletenv.StatsLogger.Unsubscribe(ch)
	}

	formatParams := map[string][]string{"full": {}}
	for out := range ch {
		stats, ok := out.(*tabletenv.LogStats)
		if !ok {
			log.Errorf("Unexpected value in query logs: %#v (expecting value of type %T)", out, &tabletenv.LogStats{})
			continue
		}
		var b bytes.Buffer
		if err := stats.Logf(&b, formatParams); err != nil {
			log.Errorf("Error formatting logStats: %v", err)
			continue
		}
		if err := writer.Info(b.String()); err != nil {
			log.Errorf("Error writing to syslog: %v", err)
			continue
		}
	}
}
728109
func (s *Service) Run(ctx context.Context, commandParts []string, options options.Run) (int, error) {
	err := s.ensureImageExists(ctx, false, false)
	if err != nil {
		return -1, err
	}

	client := s.clientFactory.Create(s)

	namer, err := NewNamer(ctx, client, s.project.Name, s.name, true)
	if err != nil {
		return -1, err
	}

	configOverride := &config.ServiceConfig{Command: commandParts, Tty: !options.DisableTty, StdinOpen: !options.DisableTty}

	c, err := s.createContainer(ctx, namer, "", configOverride, true)
	if err != nil {
		return -1, err
	}

	if err := s.connectContainerToNetworks(ctx, c, true); err != nil {
		return -1, err
	}

	if options.Detached {
		logrus.Infof("%s", c.Name())
		return 0, c.Start(ctx)
	}
	return c.Run(ctx, configOverride)
}
642671
func (s *Scheduler) Run(ctx context.Context) error {
	defer close(s.doneChan)

	updates, cancel, err := store.ViewAndWatch(s.store, s.setupTasksList)
	if err != nil {
		log.G(ctx).WithError(err).Errorf("snapshot store update failed")
		return err
	}
	defer cancel()

	// Validate resource for tasks from preassigned tasks
	// do this before other tasks because preassigned tasks like
	// global service should start before other tasks
	s.processPreassignedTasks(ctx)

	// Queue all unassigned tasks before processing changes.
	s.tick(ctx)

	const (
		// commitDebounceGap is the amount of time to wait between
		// commit events to debounce them.
		commitDebounceGap = 50 * time.Millisecond
		// maxLatency is a time limit on the debouncing.
		maxLatency = time.Second
	)
	var (
		debouncingStarted     time.Time
		commitDebounceTimer   *time.Timer
		commitDebounceTimeout <-chan time.Time
	)

	tickRequired := false

	schedule := func() {
		if len(s.pendingPreassignedTasks) > 0 {
			s.processPreassignedTasks(ctx)
		}
		if tickRequired {
			s.tick(ctx)
			tickRequired = false
		}
	}

	// Watch for changes.
	for {
		select {
		case event := <-updates:
			switch v := event.(type) {
			case api.EventCreateTask:
				if s.createTask(ctx, v.Task) {
					tickRequired = true
				}
			case api.EventUpdateTask:
				if s.updateTask(ctx, v.Task) {
					tickRequired = true
				}
			case api.EventDeleteTask:
				if s.deleteTask(v.Task) {
					// deleting tasks may free up node resource, pending tasks should be re-evaluated.
					tickRequired = true
				}
			case api.EventCreateNode:
				s.createOrUpdateNode(v.Node)
				tickRequired = true
			case api.EventUpdateNode:
				s.createOrUpdateNode(v.Node)
				tickRequired = true
			case api.EventDeleteNode:
				s.nodeSet.remove(v.Node.ID)
			case state.EventCommit:
				if commitDebounceTimer != nil {
					if time.Since(debouncingStarted) > maxLatency {
						commitDebounceTimer.Stop()
						commitDebounceTimer = nil
						commitDebounceTimeout = nil
						schedule()
					} else {
						commitDebounceTimer.Reset(commitDebounceGap)
					}
				} else {
					commitDebounceTimer = time.NewTimer(commitDebounceGap)
					commitDebounceTimeout = commitDebounceTimer.C
					debouncingStarted = time.Now()
				}
			}
		case <-commitDebounceTimeout:
			schedule()
			commitDebounceTimer = nil
			commitDebounceTimeout = nil
		case <-s.stopChan:
			return nil
		}
	}
}
993348
func (n *Server) Run() {
	if err := server.RemoveUnixSocket(n.nccSocket); err != nil {
		log.Fatalf("Failed to remove socket: %v", err)
	}
	ln, err := net.Listen("unix", n.nccSocket)
	if err != nil {
		log.Fatalf("listen error: %v", err)
	}
	defer ln.Close()
	defer os.Remove(n.nccSocket)

	seesawNCC := rpc.NewServer()
	seesawNCC.Register(&SeesawNCC{})
	go server.RPCAccept(ln, seesawNCC)

	<-n.shutdown
}
927102
func (s3 *S3) run(ctx context.Context, req *request, resp interface{}) (*http.Response, error) {
	if debug {
		log.Printf("Running S3 request: %#v", req)
	}

	u, err := req.url()
	if err != nil {
		return nil, err
	}

	hreq := &http.Request{
		URL:        u,
		Method:     req.method,
		ProtoMajor: 1,
		ProtoMinor: 1,
		Close:      true,
		Header:     req.headers,
	}

	if ctx != nil {
		hreq = hreq.WithContext(ctx)
	}

	if v, ok := req.headers["Content-Length"]; ok {
		hreq.ContentLength, _ = strconv.ParseInt(v[0], 10, 64)
		delete(req.headers, "Content-Length")
	}
	if req.payload != nil {
		hreq.Body = ioutil.NopCloser(req.payload)
	}

	if s3.client == nil {
		s3.client = &http.Client{
			Transport: &http.Transport{
				Dial: func(netw, addr string) (c net.Conn, err error) {
					c, err = net.DialTimeout(netw, addr, s3.ConnectTimeout)
					if err != nil {
						return
					}

					var deadline time.Time
					if s3.RequestTimeout > 0 {
						deadline = time.Now().Add(s3.RequestTimeout)
						c.SetDeadline(deadline)
					}

					if s3.ReadTimeout > 0 || s3.WriteTimeout > 0 {
						c = &ioTimeoutConn{
							TCPConn:         c.(*net.TCPConn),
							readTimeout:     s3.ReadTimeout,
							writeTimeout:    s3.WriteTimeout,
							requestDeadline: deadline,
						}
					}
					return
				},
			},
		}
	}

	hresp, err := s3.client.Do(hreq)
	if err != nil {
		return nil, err
	}
	if debug {
		dump, _ := httputil.DumpResponse(hresp, true)
		log.Printf("} -> %s\n", dump)
	}
	if hresp.StatusCode != 200 && hresp.StatusCode != 204 && hresp.StatusCode != 206 {
		defer hresp.Body.Close()
		return nil, buildError(hresp)
	}
	if resp != nil {
		err = xml.NewDecoder(hresp.Body).Decode(resp)
		hresp.Body.Close()
		if debug {
			log.Printf("goamz.s3> decoded xml into %#v", resp)
		}
	}
	return hresp, err
}
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)
		}()
	}
}
1086772
func (srv *Server) Run() {
	s := grpc.NewServer()
	go srv.listenForPolicyCountChanges()
	healthpb.RegisterHealthServer(s, srv)
	api.RegisterCollectorSetControllerServer(s, srv)
	reflection.Register(s)

	// Start the gRPC server.
	lis, err := net.Listen("tcp", fmt.Sprintf(":%s", "50000"))
	if err != nil {
		log.Fatalf("Failed to listen: %v", err)
	}
	s.Serve(lis) // nolint: errcheck, gosec
}
666394
func (x *xdsBalancer) run() {
	for {
		select {
		case update := <-x.grpcUpdate:
			x.handleGRPCUpdate(update)
		case update := <-x.xdsClientUpdate:
			x.handleXDSClientUpdate(update)
		case <-x.timer.C: // x.timer.C will block if we are not in fallback monitoring stage.
			x.switchFallback()
		case <-x.noSubConnAlert: // x.noSubConnAlert will block if we are not in fallback monitoring stage.
			x.switchFallback()
		case <-x.ctx.Done():
			if x.client != nil {
				x.client.close()
			}
			if x.xdsLB != nil {
				x.xdsLB.Close()
			}
			if x.fallbackLB != nil {
				x.fallbackLB.Close()
			}
			return
		}
	}
}
907592
func (cnc *CloudNodeController) Run(stopCh <-chan struct{}) {
	defer utilruntime.HandleCrash()

	// The following loops run communicate with the APIServer with a worst case complexity
	// of O(num_nodes) per cycle. These functions are justified here because these events fire
	// very infrequently. DO NOT MODIFY this to perform frequent operations.

	// Start a loop to periodically update the node addresses obtained from the cloud
	wait.Until(cnc.UpdateNodeStatus, cnc.nodeStatusUpdateFrequency, stopCh)
}
541268
func (e *EmptyService) Run(ctx context.Context, commandParts []string, options options.Run) (int, error) {
	return 0, nil
}
431590
func (d *commandSQLAll) Run() error {
	sqls, indexes := getDbCreateSQL(d.al)
	var all []string
	for i, mi := range modelCache.allOrdered() {
		queries := []string{sqls[i]}
		for _, idx := range indexes[mi.table] {
			queries = append(queries, idx.SQL)
		}
		sql := strings.Join(queries, "\n")
		all = append(all, sql)
	}
	fmt.Println(strings.Join(all, "\n\n"))

	return nil
}