// startStream assumes that getTablet() was succesfully called before and now
// tries to connect to the set tablet and start the streaming query.
// If the method returns an error, the first return value specifies if it is
// okay to retry.
162848
func (r *RestartableResultReader) startStream() (bool, error) {
	// Start the streaming query.
	r.generateQuery()
	var stream sqltypes.ResultStream

	if r.txID == 0 {
		stream = queryservice.ExecuteWithStreamer(r.ctx, r.conn, &querypb.Target{
			Keyspace:   r.tablet.Keyspace,
			Shard:      r.tablet.Shard,
			TabletType: r.tablet.Type,
		}, r.query, make(map[string]*querypb.BindVariable), nil)
	} else {
		stream = queryservice.ExecuteWithTransactionalStreamer(r.ctx, r.conn, &querypb.Target{
			Keyspace:   r.tablet.Keyspace,
			Shard:      r.tablet.Shard,
			TabletType: r.tablet.Type,
		}, r.query, make(map[string]*querypb.BindVariable), r.txID, nil)
	}

	// Read the fields information.
	cols, err := stream.Recv()
	if err != nil {
		return true /* retryable */, vterrors.Wrapf(err, "cannot read Fields for query '%v'", r.query)
	}
	r.fields = cols.Fields
	r.output = stream

	alias := topoproto.TabletAliasString(r.tablet.Alias)
	statsStreamingQueryCounters.Add(alias, 1)
	log.V(2).Infof("tablet=%v table=%v chunk=%v: Starting to stream rows using query '%v'.", alias, r.td.Name, r.chunk, r.query)
	return false, nil
}
778288
func (ts *Server) GetTablet(ctx context.Context, alias *topodatapb.TabletAlias) (*TabletInfo, error) {
	conn, err := ts.ConnForCell(ctx, alias.Cell)
	if err != nil {
		return nil, err
	}

	span, ctx := trace.NewSpan(ctx, "TopoServer.GetTablet")
	span.Annotate("tablet", topoproto.TabletAliasString(alias))
	defer span.Finish()

	tabletPath := path.Join(TabletsPath, topoproto.TabletAliasString(alias), TabletFile)
	data, version, err := conn.Get(ctx, tabletPath)
	if err != nil {
		return nil, err
	}
	tablet := &topodatapb.Tablet{}
	if err := proto.Unmarshal(data, tablet); err != nil {
		return nil, err
	}

	return &TabletInfo{
		version: version,
		Tablet:  tablet,
	}, nil
}
730605
func (qe *QueryEngine) GetStreamPlan(sql string) (*TabletPlan, error) {
	qe.mu.RLock()
	defer qe.mu.RUnlock()
	splan, err := planbuilder.BuildStreaming(sql, qe.tables)
	if err != nil {
		return nil, err
	}
	plan := &TabletPlan{Plan: splan}
	plan.Rules = qe.queryRuleSources.FilterByPlan(sql, plan.PlanID, plan.TableName().String())
	plan.buildAuthorized()
	return plan, nil
}
1017289
func (tsv *TabletServer) Begin(ctx context.Context, target *querypb.Target, options *querypb.ExecuteOptions) (transactionID int64, err error) {
	err = tsv.execRequest(
		ctx, tsv.BeginTimeout.Get(),
		"Begin", "begin", nil,
		target, options, true /* isBegin */, false, /* allowOnShutdown */
		func(ctx context.Context, logStats *tabletenv.LogStats) error {
			startTime := time.Now()
			if tsv.txThrottler.Throttle() {
				// TODO(erez): I think this should be RESOURCE_EXHAUSTED.
				return vterrors.Errorf(vtrpcpb.Code_UNAVAILABLE, "Transaction throttled")
			}
			var beginSQL string
			transactionID, beginSQL, err = tsv.teCtrl.Begin(ctx, options)
			logStats.TransactionID = transactionID

			// Record the actual statements that were executed in the logStats.
			// If nothing was actually executed, don't count the operation in
			// the tablet metrics, and clear out the logStats Method so that
			// handlePanicAndSendLogStats doesn't log the no-op.
			logStats.OriginalSQL = beginSQL
			if beginSQL != "" {
				tabletenv.QueryStats.Record("BEGIN", startTime)
			} else {
				logStats.Method = ""
			}
			return err
		},
	)
	return transactionID, err
}
1091329
func (es *EventStream) Restart() {
	es.Stop()
	es.Resume = true
	es.Start()
}
781772
func (wr *Wrangler) RefreshTabletsByShard(ctx context.Context, si *topo.ShardInfo, tabletTypes []topodatapb.TabletType, cells []string) error {
	wr.Logger().Infof("RefreshTabletsByShard called on shard %v/%v", si.Keyspace(), si.ShardName())
	tabletMap, err := wr.ts.GetTabletMapForShardByCell(ctx, si.Keyspace(), si.ShardName(), cells)
	switch {
	case err == nil:
		// keep going
	case topo.IsErrType(err, topo.PartialResult):
		wr.Logger().Warningf("RefreshTabletsByShard: got partial result for shard %v/%v, may not refresh all tablets everywhere", si.Keyspace(), si.ShardName())
	default:
		return err
	}

	// ignore errors in this phase
	wg := sync.WaitGroup{}
	for _, ti := range tabletMap {
		if tabletTypes != nil && !topoproto.IsTypeInList(ti.Type, tabletTypes) {
			continue
		}
		if ti.Hostname == "" {
			// The tablet is not running, we don't have the host
			// name to connect to, so we just skip this tablet.
			wr.Logger().Infof("Tablet %v has no hostname, skipping its RefreshState", ti.AliasString())
			continue
		}

		wg.Add(1)
		go func(ti *topo.TabletInfo) {
			wr.Logger().Infof("Calling RefreshState on tablet %v", ti.AliasString())
			// Setting an upper bound timeout to fail faster in case of an error.
			// Using 60 seconds because RefreshState should not take more than 30 seconds.
			// (RefreshState will restart the tablet's QueryService and most time will be spent on the shutdown, i.e. waiting up to 30 seconds on transactions (see Config.TransactionTimeout)).
			ctx, cancel := context.WithTimeout(ctx, 60*time.Second)
			if err := wr.tmc.RefreshState(ctx, ti.Tablet); err != nil {
				wr.Logger().Warningf("RefreshTabletsByShard: failed to refresh %v: %v", ti.AliasString(), err)
			}
			cancel()
			wg.Done()
		}(ti)
	}
	wg.Wait()

	return nil
}
245313
func (o *Outbound) CallStream(ctx context.Context, request *transport.StreamRequest) (*transport.ClientStream, error) {
	if err := o.once.WaitUntilRunning(ctx); err != nil {
		return nil, err
	}
	return o.stream(ctx, request, time.Now())
}
277591
func (conn *gRPCQueryClient) Begin(ctx context.Context, target *querypb.Target, options *querypb.ExecuteOptions) (transactionID int64, err error) {
	conn.mu.RLock()
	defer conn.mu.RUnlock()
	if conn.cc == nil {
		return 0, tabletconn.ConnClosed
	}

	req := &querypb.BeginRequest{
		Target:            target,
		EffectiveCallerId: callerid.EffectiveCallerIDFromContext(ctx),
		ImmediateCallerId: callerid.ImmediateCallerIDFromContext(ctx),
		Options:           options,
	}
	br, err := conn.c.Begin(ctx, req)
	if err != nil {
		return 0, tabletconn.ErrorFromGRPC(err)
	}
	return br.TransactionId, nil
}
707770
func (p *StatusTable) Start(name string) {
	p.Lock()
	defer p.Unlock()

	p.pool[name] = true
}
1070441
func (conn *gRPCQueryClient) UpdateStream(ctx context.Context, target *querypb.Target, position string, timestamp int64, callback func(*querypb.StreamEvent) error) error {
	// Please see comments in StreamExecute to see how this works.
	ctx, cancel := context.WithCancel(ctx)
	defer cancel()

	stream, err := func() (queryservicepb.Query_UpdateStreamClient, error) {
		conn.mu.RLock()
		defer conn.mu.RUnlock()
		if conn.cc == nil {
			return nil, tabletconn.ConnClosed
		}

		req := &querypb.UpdateStreamRequest{
			Target:            target,
			EffectiveCallerId: callerid.EffectiveCallerIDFromContext(ctx),
			ImmediateCallerId: callerid.ImmediateCallerIDFromContext(ctx),
			Position:          position,
			Timestamp:         timestamp,
		}
		stream, err := conn.c.UpdateStream(ctx, req)
		if err != nil {
			return nil, tabletconn.ErrorFromGRPC(err)
		}
		return stream, nil
	}()
	if err != nil {
		return err
	}
	for {
		r, err := stream.Recv()
		if err != nil {
			return tabletconn.ErrorFromGRPC(err)
		}
		if err := callback(r.Event); err != nil {
			if err == nil || err == io.EOF {
				return nil
			}
			return err
		}
	}
}
501686
func (r *RestartableResultReader) Next() (*sqltypes.Result, error) {
	result, err := r.output.Recv()
	if err != nil && err != io.EOF {
		// We start the retries only on the second attempt to avoid the cost
		// of starting a timer (for the retry timeout) for every Next() call
		// when no error occurs.
		alias := topoproto.TabletAliasString(r.tablet.Alias)
		statsStreamingQueryErrorsCounters.Add(alias, 1)
		log.V(2).Infof("tablet=%v table=%v chunk=%v: Failed to read next rows from active streaming query. Trying to restart stream on the same tablet. Original Error: %v", alias, r.td.Name, r.chunk, err)
		result, err = r.nextWithRetries()
	}
	if result != nil && len(result.Rows) > 0 {
		r.lastRow = result.Rows[len(result.Rows)-1]
	}
	return result, err
}
180515
func (cli *Client) StreamrouterStreamprocsGet(appname string) (*pb.StreamrouterStreamprocsReply, error) {
	req := &pb.StreamrouterStreamprocsRequest{Appname: appname}
	ctx, cancel := context.WithTimeout(context.Background(), cli.timeout)
	defer cancel()
	rpl, err := cli.streamrouterStreamprocsClient.Get(ctx, req)
	return rpl, err
}
834789
func (f StreamOutboundFunc) CallStream(ctx context.Context, request *transport.StreamRequest, out transport.StreamOutbound) (*transport.ClientStream, error) {
	return f(ctx, request, out)
}
533060
func (es *EventStream) Start() {
	if es.Events == nil {
		es.Events = make(chan []Event)
	}

	// register eventstream in the local registry for later lookup
	// in C callback
	cbInfo := registry.Add(es)
	es.registryID = cbInfo
	es.uuid = GetDeviceUUID(es.Device)
	es.start(es.Paths, cbInfo)
}
1362813
func (c *SQLiteConn) Begin() (driver.Tx, error) {
	return c.begin(context.Background())
}
282275
func (ff *firstRowFunction) GetStreamResult() (d types.Datum) {
	if ff.streamCtx == nil {
		return
	}
	d = ff.streamCtx.Value
	ff.streamCtx = nil
	return
}
1062071
func (s *session) start(ctx context.Context, description *api.NodeDescription) error {
	log.G(ctx).Debugf("(*session).start")

	errChan := make(chan error, 1)
	var (
		msg    *api.SessionMessage
		stream api.Dispatcher_SessionClient
		err    error
	)
	// Note: we don't defer cancellation of this context, because the
	// streaming RPC is used after this function returned. We only cancel
	// it in the timeout case to make sure the goroutine completes.

	// We also fork this context again from the `run` context, because on
	// `dispatcherRPCTimeout`, we want to cancel establishing a session and
	// return an error.  If we cancel the `run` context instead of forking,
	// then in `run` it's possible that we just terminate the function because
	// `ctx` is done and hence fail to propagate the timeout error to the agent.
	// If the error is not propogated to the agent, the agent will not close
	// the session or rebuild a new session.
	sessionCtx, cancelSession := context.WithCancel(ctx) // nolint: vet

	// Need to run Session in a goroutine since there's no way to set a
	// timeout for an individual Recv call in a stream.
	go func() {
		client := api.NewDispatcherClient(s.conn.ClientConn)

		stream, err = client.Session(sessionCtx, &api.SessionRequest{
			Description: description,
			SessionID:   s.sessionID,
		})
		if err != nil {
			errChan <- err
			return
		}

		msg, err = stream.Recv()
		errChan <- err
	}()

	select {
	case err := <-errChan:
		if err != nil {
			return err // nolint: vet
		}
	case <-time.After(dispatcherRPCTimeout):
		cancelSession()
		return errors.New("session initiation timed out")
	}

	s.sessionID = msg.SessionID
	s.session = stream

	return s.handleSessionMessage(ctx, msg)
}