containerd通信机制

package main

import (
	"crypto"
	"fmt"
	"os"

	"github.com/containerd/containerd/v2/cmd/containerd/command"
	"github.com/containerd/containerd/v2/internal/hasher"

	_ "github.com/containerd/containerd/v2/cmd/containerd/builtins"
)

func init() {
	crypto.RegisterHash(crypto.SHA256, hasher.NewSHA256)
}

func main() {
	app := command.App()
	if err := app.Run(os.Args); err != nil {
		fmt.Fprintf(os.Stderr, "containerd: %s\n", err)
		os.Exit(1)
	}
}

// App returns a *cli.App instance.
func App() *cli.App {
	app := cli.NewApp()
	app.Name = "containerd"
	app.Version = version.Version
	app.Usage = usage
	app.Flags = append(app.Flags, serviceFlags()...)
	app.Commands = []*cli.Command{
		configCommand,
		publishCommand,
		ociHook,
	}
	app.Action = func(cliContext *cli.Context) error {
		var (
			start       = time.Now()
			signals     = make(chan os.Signal, 2048)
			serverC     = make(chan *server.Server, 1)
			ctx, cancel = context.WithCancel(cliContext.Context)
			config      = defaultConfig()
		)

//···此处省略部分代码

		if config.Debug.Address != "" {
			var l net.Listener
			if isLocalAddress(config.Debug.Address) {
				if l, err = sys.GetLocalListener(config.Debug.Address, config.Debug.UID, config.Debug.GID); err != nil {
					return fmt.Errorf("failed to get listener for debug endpoint: %w", err)
				}
			} else {
				if l, err = net.Listen("tcp", config.Debug.Address); err != nil {
					return fmt.Errorf("failed to get listener for debug endpoint: %w", err)
				}
			}
			serve(ctx, l, server.ServeDebug)
		}
		if config.Metrics.Address != "" {
			l, err := net.Listen("tcp", config.Metrics.Address)
			if err != nil {
				return fmt.Errorf("failed to get listener for metrics endpoint: %w", err)
			}
			serve(ctx, l, server.ServeMetrics)
		}
		// setup the ttrpc endpoint
		tl, err := sys.GetLocalListener(config.TTRPC.Address, config.TTRPC.UID, config.TTRPC.GID)
		if err != nil {
			return fmt.Errorf("failed to get listener for main ttrpc endpoint: %w", err)
		}
		serve(ctx, tl, server.ServeTTRPC)

		if config.GRPC.TCPAddress != "" {
			l, err := net.Listen("tcp", config.GRPC.TCPAddress)
			if err != nil {
				return fmt.Errorf("failed to get listener for TCP grpc endpoint: %w", err)
			}
			serve(ctx, l, server.ServeTCP)
		}
		// setup the main grpc endpoint
		l, err := sys.GetLocalListener(config.GRPC.Address, config.GRPC.UID, config.GRPC.GID)
		if err != nil {
			return fmt.Errorf("failed to get listener for main endpoint: %w", err)
		}
		serve(ctx, l, server.ServeGRPC)

		readyC := make(chan struct{})
		go func() {
			server.Wait()
			close(readyC)
		}()
        
		return nil
	}
	return app
}

func serve(ctx context.Context, l net.Listener, serveFunc func(net.Listener) error) {
	path := l.Addr().String()
	log.G(ctx).WithField("address", path).Info("serving...")
	go func() {
		defer l.Close()
		if err := serveFunc(l); err != nil {
			log.G(ctx).WithError(err).WithField("address", path).Fatal("serve failure")
		}
	}()
}
//···此处省略部分代码

package sys

import (
	"fmt"
	"net"
	"os"
	"path/filepath"

	"golang.org/x/sys/unix"
)

// CreateUnixSocket creates a unix socket and returns the listener
func CreateUnixSocket(path string) (net.Listener, error) {
	// BSDs have a 104 limit
	if len(path) > 104 {
		return nil, fmt.Errorf("%q: unix socket path too long (> 104)", path)
	}
	if err := os.MkdirAll(filepath.Dir(path), 0660); err != nil {
		return nil, err
	}
	if err := unix.Unlink(path); err != nil && !os.IsNotExist(err) {
		return nil, err
	}
	return net.Listen("unix", path)
}

// GetLocalListener returns a listener out of a unix socket.
func GetLocalListener(path string, uid, gid int) (net.Listener, error) {
	// Ensure parent directory is created
	if err := mkdirAs(filepath.Dir(path), uid, gid); err != nil {
		return nil, err
	}

	l, err := CreateUnixSocket(path)
	if err != nil {
		return l, err
	}

	if err := os.Chmod(path, 0660); err != nil {
		l.Close()
		return nil, err
	}

	if err := os.Chown(path, uid, gid); err != nil {
		l.Close()
		return nil, err
	}

	return l, nil
}

func mkdirAs(path string, uid, gid int) error {
	if _, err := os.Stat(path); !os.IsNotExist(err) {
		return err
	}

	if err := os.MkdirAll(path, 0770); err != nil {
		return err
	}

	return os.Chown(path, uid, gid)
}

func CreateUnixSocket(path string) (net.Listener, error) {
	// BSDs have a 104 limit
	if len(path) > 104 {
		return nil, fmt.Errorf("%q: unix socket path too long (> 104)", path)
	}
	if err := os.MkdirAll(filepath.Dir(path), 0660); err != nil {
		return nil, err
	}
	if err := unix.Unlink(path); err != nil && !os.IsNotExist(err) {
		return nil, err
	}
	return net.Listen("unix", path)
}

func GetLocalListener(path string, uid, gid int) (net.Listener, error) {
	// Ensure parent directory is created
	if err := mkdirAs(filepath.Dir(path), uid, gid); err != nil {
		return nil, err
	}

	l, err := CreateUnixSocket(path)
	if err != nil {
		return l, err
	}

	if err := os.Chmod(path, 0660); err != nil {
		l.Close()
		return nil, err
	}

	if err := os.Chown(path, uid, gid); err != nil {
		l.Close()
		return nil, err
	}

	return l, nil
}

func mkdirAs(path string, uid, gid int) error {
	if _, err := os.Stat(path); !os.IsNotExist(err) {
		return err
	}

	if err := os.MkdirAll(path, 0770); err != nil {
		return err
	}

	return os.Chown(path, uid, gid)
}

type Client struct {
	services
	connMu    sync.Mutex
	conn      *grpc.ClientConn
	runtime   string
	defaultns string
	platform  platforms.MatchComparer
	connector func() (*grpc.ClientConn, error)
}

func New(address string, opts ...Opt) (*Client, error) {
	var copts clientOpts
	for _, o := range opts {
		if err := o(&copts); err != nil {
			return nil, err
		}
	}
	if copts.timeout == 0 {
		copts.timeout = 10 * time.Second
	}

	c := &Client{
		defaultns: copts.defaultns,
	}

	if copts.defaultRuntime != "" {
		c.runtime = copts.defaultRuntime
	} else {
		c.runtime = defaults.DefaultRuntime
	}

	if copts.defaultPlatform != nil {
		c.platform = copts.defaultPlatform
	} else {
		c.platform = platforms.Default()
	}

	if copts.services != nil {
		c.services = *copts.services
	}

	if address != "" {
		backoffConfig := backoff.DefaultConfig
		backoffConfig.MaxDelay = copts.timeout
		connParams := grpc.ConnectParams{
			Backoff: backoffConfig,
		}
		gopts := []grpc.DialOption{
			grpc.WithTransportCredentials(insecure.NewCredentials()),
			grpc.WithConnectParams(connParams),
			grpc.WithContextDialer(dialer.ContextDialer),
		}
		if len(copts.dialOptions) > 0 {
			gopts = copts.dialOptions
		}
		gopts = append(gopts, grpc.WithDefaultCallOptions(
			grpc.MaxCallRecvMsgSize(defaults.DefaultMaxRecvMsgSize),
			grpc.MaxCallSendMsgSize(defaults.DefaultMaxSendMsgSize)))
		if len(copts.callOptions) > 0 {
			gopts = append(gopts, grpc.WithDefaultCallOptions(copts.callOptions...))
		}
		if copts.defaultns != "" {
			unary, stream := newNSInterceptors(copts.defaultns)
			gopts = append(gopts, grpc.WithChainUnaryInterceptor(unary))
			gopts = append(gopts, grpc.WithChainStreamInterceptor(stream))
		}

		connector := func() (*grpc.ClientConn, error) {
			conn, err := grpc.NewClient(dialer.DialAddress(address), gopts...) //gRPC连接
			if err != nil {
				return nil, fmt.Errorf("failed to dial %q: %w", address, err)
			}
			return conn, nil
		}
		conn, err := connector()
		if err != nil {
			return nil, err
		}
		c.conn, c.connector = conn, connector
	}

	if copts.services == nil && c.conn == nil {
		return nil, fmt.Errorf("no grpc connection or services is available: %w", errdefs.ErrUnavailable)
	}

	// check namespace labels for default runtime
	if copts.defaultRuntime == "" && c.defaultns != "" {
		if label, err := c.GetLabel(context.Background(), defaults.DefaultRuntimeNSLabel); err != nil {
			return nil, err
		} else if label != "" {
			c.runtime = label
		}
	}

	return c, nil
}

func NewWithConn(conn *grpc.ClientConn, opts ...Opt) (*Client, error) {
	var copts clientOpts
	for _, o := range opts {
		if err := o(&copts); err != nil {
			return nil, err
		}
	}
	c := &Client{
		defaultns: copts.defaultns,
		conn:      conn,
		runtime:   defaults.DefaultRuntime,
	}

	if copts.defaultPlatform != nil {
		c.platform = copts.defaultPlatform
	} else {
		c.platform = platforms.Default()
	}

	// check namespace labels for default runtime
	if copts.defaultRuntime == "" && c.defaultns != "" {
		if label, err := c.GetLabel(context.Background(), defaults.DefaultRuntimeNSLabel); err != nil {
			return nil, err
		} else if label != "" {
			c.runtime = label
		}
	}

	if copts.services != nil {
		c.services = *copts.services
	}
	return c, nil
}

func (c *Client) Reconnect() error {
	if c.connector == nil {
		return fmt.Errorf("unable to reconnect to containerd, no connector available: %w", errdefs.ErrUnavailable)
	}
	c.connMu.Lock()
	defer c.connMu.Unlock()
	c.conn.Close()
	conn, err := c.connector()
	if err != nil {
		return err
	}
	c.conn = conn
	return nil
}

func (c *Client) IsServing(ctx context.Context) (bool, error) {
	c.connMu.Lock()
	if c.conn == nil {
		c.connMu.Unlock()
		return false, fmt.Errorf("no grpc connection available: %w", errdefs.ErrUnavailable)
	}
	c.connMu.Unlock()
	r, err := c.HealthService().Check(ctx, &grpc_health_v1.HealthCheckRequest{}, grpc.WaitForReady(true))
	if err != nil {
		return false, err
	}
	return r.Status == grpc_health_v1.HealthCheckResponse_SERVING, nil
}

func NewClient(target string, opts ...DialOption) (conn *ClientConn, err error) {
	cc := &ClientConn{
		target: target,
		conns:  make(map[*addrConn]struct{}),
		dopts:  defaultDialOptions(),
	}

	cc.retryThrottler.Store((*retryThrottler)(nil))
	cc.safeConfigSelector.UpdateConfigSelector(&defaultConfigSelector{nil})
	cc.ctx, cc.cancel = context.WithCancel(context.Background())

	// Apply dial options.
	disableGlobalOpts := false
	for _, opt := range opts {
		if _, ok := opt.(*disableGlobalDialOptions); ok {
			disableGlobalOpts = true
			break
		}
	}

	if !disableGlobalOpts {
		for _, opt := range globalDialOptions {
			opt.apply(&cc.dopts)
		}
	}

	for _, opt := range opts {
		opt.apply(&cc.dopts)
	}

	// Determine the resolver to use.
	if err := cc.initParsedTargetAndResolverBuilder(); err != nil {
		return nil, err
	}

	for _, opt := range globalPerTargetDialOptions {
		opt.DialOptionForTarget(cc.parsedTarget.URL).apply(&cc.dopts)
	}

	chainUnaryClientInterceptors(cc)
	chainStreamClientInterceptors(cc)

	if err := cc.validateTransportCredentials(); err != nil {
		return nil, err
	}

	if cc.dopts.defaultServiceConfigRawJSON != nil {
		scpr := parseServiceConfig(*cc.dopts.defaultServiceConfigRawJSON, cc.dopts.maxCallAttempts)
		if scpr.Err != nil {
			return nil, fmt.Errorf("%s: %v", invalidDefaultServiceConfigErrPrefix, scpr.Err)
		}
		cc.dopts.defaultServiceConfig, _ = scpr.Config.(*ServiceConfig)
	}
	cc.mkp = cc.dopts.copts.KeepaliveParams

	if err = cc.initAuthority(); err != nil {
		return nil, err
	}

	// Register ClientConn with channelz. Note that this is only done after
	// channel creation cannot fail.
	cc.channelzRegistration(target)
	channelz.Infof(logger, cc.channelz, "parsed dial target is: %#v", cc.parsedTarget)
	channelz.Infof(logger, cc.channelz, "Channel authority set to %q", cc.authority)

	cc.csMgr = newConnectivityStateManager(cc.ctx, cc.channelz)
	cc.pickerWrapper = newPickerWrapper(cc.dopts.copts.StatsHandlers)

	cc.initIdleStateLocked() // Safe to call without the lock, since nothing else has a reference to cc.
	cc.idlenessMgr = idle.NewManager((*idler)(cc), cc.dopts.idleTimeout)
	return cc, nil
}

func NewClient(target string, opts ...DialOption) (*ClientConn, error) {
    cc := &ClientConn{
        target: target,
        conns:  make(map[*addrConn]struct{}),
        dopts:  defaultDialOptions(),
    }
    
    // 解析 target 地址，选择合适的 resolver (解析器)
    if err := cc.initParsedTargetAndResolverBuilder(); err != nil {
        return nil, err
    }

    // 初始化连接管理和负载均衡
    cc.csMgr = newConnectivityStateManager(cc.ctx, cc.channelz)
    cc.pickerWrapper = newPickerWrapper(cc.dopts.copts.StatsHandlers)


    return cc, nil
}

func invoke(ctx context.Context, method string, req, reply any, cc *ClientConn, opts ...CallOption) error {
	cs, err := newClientStream(ctx, unaryStreamDesc, cc, method, opts...)
	if err != nil {
		return err
	}
	if err := cs.SendMsg(req); err != nil {
		return err
	}
	return cs.RecvMsg(reply)
}

func newClientStream(ctx context.Context, desc *StreamDesc, cc *ClientConn, method string, opts ...CallOption) (_ ClientStream, err error) {
    // 初始化 stream，设置传输层
    t, err := cc.getTransport(ctx, opts...)
    if err != nil {
        return nil, err
    }
    
    // 开始流
    s, err := t.NewStream(ctx, hdr)
    if err != nil {
        return nil, err
    }

    return &clientStream{s: s, desc: desc}, nil
}

func (t *http2Client) NewStream(ctx context.Context, callHdr *CallHdr) (_ *Stream, err error) {
    // 创建 HTTP/2 流
    t.mu.Lock()
    s := &Stream{
        ...
    }
    t.activeStreams[s.id] = s
    t.mu.Unlock()

    // 发送请求头部
    err = t.framer.writeHeaders(s, ...)
    if err != nil {
        return nil, err
    }
    
    return s, nil
}

//(cs *clientStream) SendMsg(m any) (err error)函数gRPC-Go 客户端用于发送消息的核心方法。它执行了从消息准备到实际发送的整个过程，并处理错误和重试逻辑。
func (cs *clientStream) SendMsg(m any) (err error) {
	//错误处理的 defer 逻辑
    defer func() {
		if err != nil && err != io.EOF {
			cs.finish(err)
		}
	}()
	if cs.sentLast {   //防止重复调用
		return status.Errorf(codes.Internal, "SendMsg called after CloseSend")
	}
	if !cs.desc.ClientStreams {   //处理非客户端流式的 RPC
		cs.sentLast = true
	}

	// load hdr, payload, data
	hdr, payload, data, err := prepareMsg(m, cs.codec, cs.cp, cs.comp)
	if err != nil {  //准备消息数据，比如编码什么的
		return err
	}

	// TODO(dfawley): should we be checking len(data) instead?
	if len(payload) > *cs.callInfo.maxSendMessageSize {  //检查消息大小
		return status.Errorf(codes.ResourceExhausted, "trying to send message larger than max (%d vs. %d)", len(payload), *cs.callInfo.maxSendMessageSize)
	}
	op := func(a *csAttempt) error {   //消息发送逻辑和重试机制
		return a.sendMsg(m, hdr, payload, data)
	}
	err = cs.withRetry(op, func() { cs.bufferForRetryLocked(len(hdr)+len(payload), op) })
    
	if len(cs.binlogs) != 0 && err == nil {  //二进制日志记录
		cm := &binarylog.ClientMessage{
			OnClientSide: true,
			Message:      data,
		}
		for _, binlog := range cs.binlogs {  
			binlog.Log(cs.ctx, cm)
		}
	}
	return err
}

//func (cs *clientStream) RecvMsg(m any) error 是 gRPC-Go 客户端用于接收服务器响应消息的核心方法。这个函数在客户端接收服务器的响应时执行，并包括了错误处理、重试机制、以及日志记录等功能。
func (cs *clientStream) RecvMsg(m any) error {
    //二进制日志记录初始化
	if len(cs.binlogs) != 0 && !cs.serverHeaderBinlogged {  
		// Call Header() to binary log header if it's not already logged.
		cs.Header()
	}
    //接收信息的初始化
	var recvInfo *payloadInfo
	if len(cs.binlogs) != 0 {
		recvInfo = &payloadInfo{}
	}
    //消息接收逻辑及重试机制
	err := cs.withRetry(func(a *csAttempt) error {
		return a.recvMsg(m, recvInfo)
	}, cs.commitAttemptLocked)
    //二进制日志记录处理
	if len(cs.binlogs) != 0 && err == nil {
		sm := &binarylog.ServerMessage{
			OnClientSide: true,
			Message:      recvInfo.uncompressedBytes,
		}
		for _, binlog := range cs.binlogs {
			binlog.Log(cs.ctx, sm)
		}
	}
    // 结束流或处理错误
	if err != nil || !cs.desc.ServerStreams {
		// err != nil or non-server-streaming indicates end of stream.
		cs.finish(err)
	}
	return err
}

//sendMsg 函数的主要职责是将编码后的消息通过底层传输层发送给服务器，并处理发送过程中的错误和统计信息。
func (a *csAttempt) sendMsg(m any, hdr, payld, data []byte) error {
	cs := a.cs
	if a.trInfo != nil {
		a.mu.Lock()
		if a.trInfo.tr != nil {
			a.trInfo.tr.LazyLog(&payload{sent: true, msg: m}, true)
		}
		a.mu.Unlock()
	}
	if err := a.t.Write(a.s, hdr, payld, &transport.Options{Last: !cs.desc.ClientStreams}); err != nil {
		if !cs.desc.ClientStreams {
			// For non-client-streaming RPCs, we return nil instead of EOF on error
			// because the generated code requires it.  finish is not called; RecvMsg()
			// will call it with the stream's status independently.
			return nil
		}
		return io.EOF
	}
	for _, sh := range a.statsHandlers {
		sh.HandleRPC(a.ctx, outPayload(true, m, data, payld, time.Now()))
	}
	if channelz.IsOn() {
		a.t.IncrMsgSent()
	}
	return nil
}

//recvMsg 函数的主要职责是从服务器接收消息并解码，同时处理接收过程中的错误、解压缩和统计信息。
func (a *csAttempt) recvMsg(m any, payInfo *payloadInfo) (err error) {
	cs := a.cs
	if len(a.statsHandlers) != 0 && payInfo == nil {
		payInfo = &payloadInfo{}
	}

	if !a.decompSet {
		// Block until we receive headers containing received message encoding.
		if ct := a.s.RecvCompress(); ct != "" && ct != encoding.Identity {
			if a.dc == nil || a.dc.Type() != ct {
				// No configured decompressor, or it does not match the incoming
				// message encoding; attempt to find a registered compressor that does.
				a.dc = nil
				a.decomp = encoding.GetCompressor(ct)
			}
		} else {
			// No compression is used; disable our decompressor.
			a.dc = nil
		}
		// Only initialize this state once per stream.
		a.decompSet = true
	}
	err = recv(a.p, cs.codec, a.s, a.dc, m, *cs.callInfo.maxReceiveMessageSize, payInfo, a.decomp)
	if err != nil {
		if err == io.EOF {
			if statusErr := a.s.Status().Err(); statusErr != nil {
				return statusErr
			}
			return io.EOF // indicates successful end of stream.
		}

		return toRPCErr(err)
	}
	if a.trInfo != nil {
		a.mu.Lock()
		if a.trInfo.tr != nil {
			a.trInfo.tr.LazyLog(&payload{sent: false, msg: m}, true)
		}
		a.mu.Unlock()
	}
	for _, sh := range a.statsHandlers {
		sh.HandleRPC(a.ctx, &stats.InPayload{
			Client:   true,
			RecvTime: time.Now(),
			Payload:  m,
			// TODO truncate large payload.
			Data:             payInfo.uncompressedBytes,
			WireLength:       payInfo.compressedLength + headerLen,
			CompressedLength: payInfo.compressedLength,
			Length:           len(payInfo.uncompressedBytes),
		})
	}
	if channelz.IsOn() {
		a.t.IncrMsgRecv()
	}
	if cs.desc.ServerStreams {
		// Subsequent messages should be received by subsequent RecvMsg calls.
		return nil
	}
	// Special handling for non-server-stream rpcs.
	// This recv expects EOF or errors, so we don't collect inPayload.
	err = recv(a.p, cs.codec, a.s, a.dc, m, *cs.callInfo.maxReceiveMessageSize, nil, a.decomp)
	if err == nil {
		return toRPCErr(errors.New("grpc: client streaming protocol violation: get <nil>, want <EOF>"))
	}
	if err == io.EOF {
		return a.s.Status().Err() // non-server streaming Recv returns nil on success
	}
	return toRPCErr(err)
}

ctx := stream.Context()
ctx = contextWithServer(ctx, s)
var ti *traceInfo
if EnableTracing {
	tr := newTrace("grpc.Recv."+methodFamily(stream.Method()), stream.Method())
	ctx = newTraceContext(ctx, tr)
	ti = &traceInfo{
		tr: tr,
		firstLine: firstLine{
			client:     false,
			remoteAddr: t.Peer().Addr,
		},
	}
	if dl, ok := ctx.Deadline(); ok {
		ti.firstLine.deadline = time.Until(dl)
	}
}

sm := stream.Method()
if sm != "" && sm[0] == '/' {
	sm = sm[1:]
}
pos := strings.LastIndex(sm, "/")
if pos == -1 {
	// 处理错误的请求方法
	if ti != nil {
		ti.tr.LazyLog(&fmtStringer{"Malformed method name %q", []any{sm}}, true)
		ti.tr.SetError()
	}
	errDesc := fmt.Sprintf("malformed method name: %q", stream.Method())
	if err := t.WriteStatus(stream, status.New(codes.Unimplemented, errDesc)); err != nil {
		if ti != nil {
			ti.tr.LazyLog(&fmtStringer{"%v", []any{err}}, true)
			ti.tr.SetError()
		}
		channelz.Warningf(logger, s.channelz, "grpc: Server.handleStream failed to write status: %v", err)
	}
	if ti != nil {
		ti.tr.Finish()
	}
	return
}
service := sm[:pos]
method := sm[pos+1:]

md, _ := metadata.FromIncomingContext(ctx)
for _, sh := range s.opts.statsHandlers {
	ctx = sh.TagRPC(ctx, &stats.RPCTagInfo{FullMethodName: stream.Method()})
	sh.HandleRPC(ctx, &stats.InHeader{
		FullMethod:  stream.Method(),
		RemoteAddr:  t.Peer().Addr,
		LocalAddr:   t.Peer().LocalAddr,
		Compression: stream.RecvCompress(),
		WireLength:  stream.HeaderWireLength(),
		Header:      md,
	})
}
stream.SetContext(ctx)

srv, knownService := s.services[service]
if knownService {
	if md, ok := srv.methods[method]; ok {
		s.processUnaryRPC(ctx, t, stream, srv, md, ti)
		return
	}
	if sd, ok := srv.streams[method]; ok {
		s.processStreamingRPC(ctx, t, stream, srv, sd, ti)
		return
	}
}

if unknownDesc := s.opts.unknownStreamDesc; unknownDesc != nil {
	s.processStreamingRPC(ctx, t, stream, nil, unknownDesc, ti)
	return
}
var errDesc string
if !knownService {
	errDesc = fmt.Sprintf("unknown service %v", service)
} else {
	errDesc = fmt.Sprintf("unknown method %v for service %v", method, service)
}
if ti != nil {
	ti.tr.LazyPrintf("%s", errDesc)
	ti.tr.SetError()
}
if err := t.WriteStatus(stream, status.New(codes.Unimplemented, errDesc)); err != nil {
	if ti != nil {
		ti.tr.LazyLog(&fmtStringer{"%v", []any{err}}, true)
		ti.tr.SetError()
	}
	channelz.Warningf(logger, s.channelz, "grpc: Server.handleStream failed to write status: %v", err)
}
if ti != nil {
	ti.tr.Finish()
}