go gRPC 客户端内存暴涨原因分析

客户端内存暴涨原因分析"/>

go gRPC 客户端内存暴涨原因分析

创建一个 gRPC 客户端连接，会创建的几个协程：

1）transport.loopyWriter.run 往服务端发送数据协程，流控时会阻塞，结果是数据堆积，内存上涨

2）transport.http2Client.reader 接收服务端数据协程，并会调用 t.controlBuf.throttle() 执行流控

现象描述：

客户端到服务端单个连接，压测时内存快速增长，直到 OOM 挂掉。在 OOM 之前停止压测，内存会逐渐下降。客户端到服务端改为两个连接时，压测时未出现内存快速增长。

问题原因：

每一个 gRPC 连接均有一个独立的队列，挂在该连接的所有 streams 共享，请求相当于生产，往服务端发送请求相当于消费，当生产速度大于消费速度时，就会出现内存持续上长。该队列没有长度限制，所以会持续上长。快速上涨的原因是协程 transport.loopyWriter).run 没有被调度运行，队列消费停止，导致队列只增不减。停止压测后，协程 transport.loopyWriter).run 会恢复执行。

当不再消费时，可观察到大量如下协程：

grpc/internal/transport.(*Stream).waitOnHeader (0x90c8d5)
runtime/netpoll.go:220 internal/poll.runtime_pollWait (0x46bdd5)

使用 netstat 命令可观察到发送队列大量堆积。

解决方案：

控制生产速度，即控制单个 gRPC 客户端连接发送的请求数量。此外，还可以启用客户端的 keepalive 关闭连接。

后话

go gRPC 如果提供取 controlBuffer 的队列 list 的大小接口，可使得更为简单和友好。

相关源码：

• http2Client

// 源码所在文件：google.golang/grpc/http2_client.go
// http2Client 实现了接口 ClientTransport
// http2Client implements the ClientTransport interface with HTTP2.
type http2Client struct {conn net.Conn // underlying communication channelloopy *loopyWriter // 生产和消费关联的队列在这里面，所在文件：controlbuf.go// controlBuf delivers all the control related tasks (e.g., window// updates, reset streams, and various settings) to the controller.controlBuf *controlBuffer // 所在文件：controlbuf.gomaxConcurrentStreams  uint32streamQuota           int64streamsQuotaAvailable chan struct{}waitingStreams        uint32initialWindowSize int32
}type controlBuffer struct {list *itemList // 队列
}type loopyWriter struct {// 关联上 controlBuffer，// 消费 controlBuffer 中的队列 list，// 生产由 http2Client 通过 controlBuffer 进行。cbuf *controlBuffer
}

• 一个 gRPC 客户端连接被创建时，即会创建一个 run 协程，run 协程为队列的消费者

// 源码所在文件：internal/transport/http2_client.go
// 所在包名：transport
// 打断点方法：
// (dlv) b transport.newHTTP2Client
// 被调用：协程 grpc.addrConn.resetTransport
func newHTTP2Client(connectCtx, ctx context.Context, addr resolver.Address, opts ConnectOptions, onPrefaceReceipt func(), onGoAway func(GoAwayReason), onClose func()) (_ *http2Client, err error) {// 建立连接，注意不同于 grpc.Dial，// grpc.Dial 实际不包含连接，对于 block 调用也只是等待连接状态为 Ready 。// transport.dial 的实现调用了 net.Dialer.DialContext，// 而 net.Dialer.DialContext 是更底层 Go 自带包的组成部分，不是 gRPC 的组成部分。// net.Dialer.DialContext 的实现支持：TCP、UDP、Unix等：。conn, err := dial(connectCtx, opts.Dialer, addr.Addr)t.controlBuf = newControlBuffer(t.ctxDone) // 含发送队列的初始化if t.keepaliveEnabled {t.kpDormancyCond = sync.NewCond(&t.mu)go t.keepalive() // 保活协程}// Start the reader goroutine for incoming message. Each transport has// a dedicated goroutine which reads HTTP2 frame from network. Then it// dispatches the frame to the corresponding stream entity.go t.reader()// Send connection preface to server.n, err := t.conn.Write(clientPreface)go func() {t.loopy = newLoopyWriter(clientSide, t.framer, t.controlBuf, t.bdpEst)err := t.loopy.run()}
}0  0x00000000008f305b in google.golang/grpc/internal/transport.newHTTP2Clientat /root/go/pkg/mod/google.golang/grpc@v1.33.2/internal/transport/http2_client.go:166
1  0x00000000009285a8 in google.golang/grpc/internal/transport.NewClientTransportat /root/go/pkg/mod/google.golang/grpc@v1.33.2/internal/transport/transport.go:577
2  0x00000000009285a8 in google.golang/grpc.(*addrConn).createTransportat /root/go/pkg/mod/google.golang/grpc@v1.33.2/clientconn.go:1297
3  0x0000000000927e48 in google.golang/grpc.(*addrConn).tryAllAddrsat /root/go/pkg/mod/google.golang/grpc@v1.33.2/clientconn.go:1227// 下列的 grpc.addrConn.resetTransport 是一个协程
4  0x000000000092737f in google.golang/grpc.(*addrConn).resetTransportat /root/go/pkg/mod/google.golang/grpc@v1.33.2/clientconn.go:1142
5  0x0000000000471821 in runtime.goexitat /usr/local/go/src/runtime/asm_amd64.s:1374// 源码所在文件：grpc/clientconn.go
// 所在包名：grpc
// 被调用：grpc.addrConn.getReadyTransport
func (ac *addrConn) connect() error {// Start a goroutine connecting to the server asynchronously.go ac.resetTransport()
}// 传统类型的 RPC 调用从 grpc.ClientConn.Invoke 开始：
//    XXX.pb.go // 编译 .proto 生成的文件
// -> main.helloServiceClient.Hello
// -> grpc.ClientConn.Invoke // 在 call.go 中，如果是 stream RPC，则从调用 grpc.ClientConn.NewStream 开始
// -> grpc.invoke // 在 call.go 中
// -> grpc.newClientStream // 在 stream.go 中
// -> grpc.clientStream.newAttemptLocked // 在 stream.go 中
// -> grpc.ClientConn.getTransport // 在 clientconn.go 中
// -> grpc.pickerWrapper.pick // 在 picker_wrapper.go 中
// -> grpc.addrConn.getReadyTransport
// -> grpc.addrConn.connect // 创建协程 resetTransport
// -> grpc.addrConn.resetTransport // ***是一个协程***
// -> grpc.addrConn.tryAllAddrs
// -> grpc.addrConn.createTransport // 在clientconn.go 中
// -> transport.NewClientTransport // 在 transport.go 中
// -> transport.newHTTP2Client
// -> transport.dial
// -> net.Dialer.DialContext // net 为 Go 自带包，不是 gRPC 包
// -> net.sysDialer.dialSerial
// -> net.sysDialer.dialSingle
// -> net.sysDialer.dialTCP/dialUDP/dialUnix/dialIP
// -> net.sysDialer.doDialTCP // 以 dialTCP 为例
// -> net.internetSocket // 从这开始，和 C 语言的使用类似了，只不过包装了不同平台的
// -> net.socket
// -> net.sysSocket
//
// stream 类型的 RPC 从 NewStream 开始：
// grpc.newClientStream 除被 grpc.invoke 调用外，还会被 stream.go 中的 grpc.ClientConn.NewStream 直接调用
//    XXX.pb.go // 编译 .proto 生成的文件
// -> grpc.ClientConn.NewStream // 在 stream.go 中
// -> grpc.newClientStream // 在 stream.go 中
// -> 从这开始同上述流程// 源码所在文件：grpc/clientconn.go
// 所在包名：grpc
// 被调用：调用源头为 grpc.ClientConn.NewStream，其实是 grpc.newClientStream 。
// getReadyTransport returns the transport if ac's state is READY.
// Otherwise it returns nil, false.
// If ac's state is IDLE, it will trigger ac to connect.
func (ac *addrConn) getReadyTransport() (transport.ClientTransport, bool) {// Trigger idle ac to connect.if idle {ac.connect()}
}

• 消费协程 run 相关源代码摘要

// 源码所在文件：internal/transport/controlbuf.go
// Loopy receives frames from the control buffer.
// Each frame is handled individually; most of the work done by loopy goes
// into handling data frames. Loopy maintains a queue of active streams, and each
// stream maintains a queue of data frames; as loopy receives data frames
// it gets added to the queue of the relevant stream.
// Loopy goes over this list of active streams by processing one node every iteration,
// thereby closely resemebling to a round-robin scheduling over all streams. While
// processing a stream, loopy writes out data bytes from this stream capped by the min
// of http2MaxFrameLen, connection-level flow control and stream-level flow control.
type loopyWriter struct {// cbuf 维护了队列（list *itemList），// 如果不加控制，就会导致内存大涨。cbuf      *controlBuffersendQuota uint32// estdStreams is map of all established streams that are not cleaned-up yet.// On client-side, this is all streams whose headers were sent out.// On server-side, this is all streams whose headers were received.estdStreams map[uint32]*outStream // Established streams.// activeStreams is a linked-list of all streams that have data to send and some// stream-level flow control quota.// Each of these streams internally have a list of data items(and perhaps trailers// on the server-side) to be sent out.activeStreams *outStreamList
}// 源码所在文件：internal/transport/controlbuf.go
func (l *loopyWriter) run() (err error) {// 通过 get 间接调用 dequeue 和 dequeueAllfor {it, err := l.cbuf.get(true)if err != nil {return err}if err = l.handle(it); err != nil {return err}if _, err = l.processData(); err != nil {return err}}
}func (c *controlBuffer) get(block bool) (interface{}, error) {for {c.mu.Lock() // 队列操作需要加锁保护......// 消费队列（出队）h := c.list.dequeue().(cbItem)......if !block {c.mu.Unlock()return nil, nil}// 阻塞c.consumerWaiting = truec.mu.Unlock()select {case <-c.ch: // 对应 executeAndPut 中唤醒的：c.ch <- struct{}case <-c.done:c.finish() // 清空队列return nil, ErrConnClosing // indicates that the transport is closing}}
}func (c *controlBuffer) finish() {......// 清空队列for head := c.list.dequeueAll(); head != nil; head = head.next {......
}

• 特别说明

每一次 gRPC 调用，客户端均会创建一个新的 Stream，
该特性使得同一 gRPC 连接可以同时处理多个调用。请求的发送并不是同步的，而是基于队列的异步发送。
每一个 gRPC 客户端连接均有一个自己的队列，gRPC 并没有直接限定队列大小，所以如果不加任何限制则会内存暴涨，直到 OOM 发生。

• 生产者：发起调用的客户端*

message HelloReq { // 请求string text = 1;
}
message HelloRes { // 响应string text = 1;
}
service HelloService {rpc Hello(HelloReq) returns (HelloRes) {}
}grpcClient := grpc.Dial(endpoint, opts)
helloClient := NewHelloServiceClient(grpcClient)
// Hello 调用为生产源头
res, err := helloClient.Hello(ctx, &req)// Hello 的实现，为 protoc 编译生成的代码
func (c *helloServiceClient) Hello(ctx context.Context, in *HelloReq, opts ...grpc.CallOption) (*HelloRes, error) {out := new(HelloRes)err := c.Invoke(ctx, "/main.HelloService/Hello", in, out, opts...)if err != nil {return nil, err}return out, nil
}// 源码所在文件：google.golang/grpc/call.go
func (cc *ClientConn) Invoke(ctx context.Context, method string, args, reply interface{}, opts ...CallOption) error {// allow interceptor to see all applicable call options, which means those// configured as defaults from dial option as well as per-call optionsopts = combine(cc.dopts.callOptions, opts)if cc.dopts.unaryInt != nil {return cc.dopts.unaryInt(ctx, method, args, reply, cc, invoke, opts...)}// 转调用私有的 invoke 函数return invoke(ctx, method, args, reply, cc, opts...)
}// 源码所在文件：google.golang/grpc/call.go
func invoke(ctx context.Context, method string, req, reply interface{}, cc *ClientConn, opts ...CallOption) error {// newClientStream 间接往队列中生产消息// pprof 显示 newClientStream 调用的 withRetry 占用内存大头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)
}// 源码所在文件：google.golang/grpc/stream.go
// 设置断点：(dlv) b clientStream.SendMsg
func (cs *clientStream) SendMsg(m interface{}) (err error) {
}// 源码所在文件：google.golang/grpc/stream.go
// pprof 显示 newClientStream 消耗太多内存，而这又发生在其调用的 withRetry 中
func newClientStream(ctx context.Context, desc *StreamDesc, cc *ClientConn, method string, opts ...CallOption) (_ ClientStream, err error) {// 问题出在 newStream 分配的内存越来越多，// 但并非严格的泄漏，只是不断积累，但压力下来后会缓慢释放。// newStream 的实现是调用 NewStream：// cs.callHdr.PreviousAttempts = cs.numRetries// s, err := a.t.NewStream(cs.ctx, cs.callHdr)// cs.attempt.s = s// 这里 a 的类型为 csAttempt：// implements a single transport stream attempt within a clientStreamop := func(a *csAttempt) error { return a.newStream() }// 内存问题所在：withRetry，进一步内存发生在非直接调用的：NewStreamif err := cs.withRetry(op, func() { cs.bufferForRetryLocked(0, op) }); err != nil {cs.finish(err)return nil, err}
}// 源码所在文件：google.golang/grpc/stream.go
func (cs *clientStream) withRetry(op func(a *csAttempt) error, onSuccess func()) error {for { // 循环，内存上涨，这儿并没有循环// 调用 newStream，// 这里间接往队列中生产消息。err := op(a)}
}// 源码所在文件：google.golang/grpc/stream.go
func (a *csAttempt) newStream() error {cs := a.cscs.callHdr.PreviousAttempts = cs.numRetries// 下列的 t 类型为：transport.ClientTransport，// 但注意 transport.ClientTransport 是一个 interface，并不是 struct。// 而 http2Client 是一个针对 ClientTransport 接口的实现。s, err := a.t.NewStream(cs.ctx, cs.callHdr)
}// 源码所在文件：google.golang/grpc/internal/transport/http2_client.go
// NewStream creates a stream and registers it into the transport as "active"
// streams.
func (t *http2Client) NewStream(ctx context.Context, callHdr *CallHdr) (_ *Stream, err error) {// 内存上涨，是因为队列中存了大量的 headerFields 和 sheaderFields, err := t.createHeaderFields(ctx, callHdr)s := t.newStream(ctx, callHdr)// hdr 聚合了 headerFields 和 shdr := &headerFrame{hf:        headerFields,initStream: func(id uint32) error {t.activeStreams[id] = s},wq:         s.wq,}for {// 调用 executeAndPut 入队（生产）// 内存上涨，是因为队列中存了大量的 hdr 。success, err := t.controlBuf.executeAndPut(func(it interface{}) bool {}, hdr)}
}// 源码所在文件：google.golang/grpc/internal/transport/controlbuf.go
func (c *controlBuffer) executeAndPut(f func(it interface{}) bool, it cbItem) (bool, error) {// 入队操作（生产）// 当入队快于出队（消费）时，就会出现内存上涨。c.list.enqueue(it)if it.isTransportResponseFrame() { // 调用接口 cbItem 定义的方法// counts the number of queued items that represent the response of an action initiated by the peer// 变量 transportResponseFrames 记录了队列大小c.transportResponseFrames++}
}// 源码所在文件：google.golang/grpc/internal/transport/controlbuf.go
func (c *controlBuffer) put(it cbItem) error {// 入队操作（生产）_, err := c.executeAndPut(nil, it)return err
}type cbItem interface {isTransportResponseFrame() bool
}
func (*registerStream) isTransportResponseFrame() bool { return false }
func (h *headerFrame) isTransportResponseFrame() bool {return h.cleanup != nil && h.cleanup.rst // Results in a RST_STREAM
}
func (c *cleanupStream) isTransportResponseFrame() bool { return c.rst } // Results in a RST_STREAM
func (*dataFrame) isTransportResponseFrame() bool { return false }
func (*incomingWindowUpdate) isTransportResponseFrame() bool { return false }
func (*outgoingWindowUpdate) isTransportResponseFrame() bool {return false // window updates are throttled by thresholds
}
func (*incomingSettings) isTransportResponseFrame() bool { return true } // Results in a settings ACK
func (*outgoingSettings) isTransportResponseFrame() bool { return false }
func (*incomingGoAway) isTransportResponseFrame() bool { return false }
func (*goAway) isTransportResponseFrame() bool { return false }
func (*ping) isTransportResponseFrame() bool { return true }
func (*outFlowControlSizeRequest) isTransportResponseFrame() bool { return false }

• **协程 transport.loopyWriter.run **

0  0x000000000043bfa5 in runtime.goparkat /usr/local/go/src/runtime/proc.go:307
1  0x000000000044c10f in runtime.selectgoat /usr/local/go/src/runtime/select.go:338一个连接只有一个 run 协程
2  0x00000000008eca2f in google.golang/grpc/internal/transport.(*controlBuffer).getat /root/go/pkg/mod/google.golang/grpc@v1.33.2/internal/transport/controlbuf.go:417
3  0x00000000008ed76e in google.golang/grpc/internal/transport.(*loopyWriter).runat /root/go/pkg/mod/google.golang/grpc@v1.33.2/internal/transport/controlbuf.go:544
4  0x000000000090f13b in google.golang/grpc/internal/transport.newHTTP2Client.func3at /root/go/pkg/mod/google.golang/grpc@v1.33.2/internal/transport/http2_client.go:356
5  0x0000000000471821 in runtime.goexitat /usr/local/go/src/runtime/asm_amd64.s:1374

• 协程 transport.controlBuffer.throttle
• 协程 transport.http2Client.reader

0  0x000000000043bfa5 in runtime.goparkat /usr/local/go/src/runtime/proc.go:307
1  0x000000000044c10f in runtime.selectgoat /usr/local/go/src/runtime/select.go:338
2  0x00000000008ec335 in google.golang/grpc/internal/transport.(*controlBuffer).throttleat /root/go/pkg/mod/google.golang/grpc@v1.33.2/internal/transport/controlbuf.go:319文件 http2_client.go 函数 transport.newHTTP2Client 会创建协程 reader
3  0x00000000008fdadd in google.golang/grpc/internal/transport.(*http2Client).readerat /root/go/pkg/mod/google.golang/grpc@v1.33.2/internal/transport/http2_client.go:1293
4  0x0000000000471821 in runtime.goexitat /usr/local/go/src/runtime/asm_amd64.s:1374

• 协程 transport.Stream.waitOnHeader

// 大量 waitOnHeader 协程
func (s *Stream) waitOnHeader() {if s.headerChan == nil {// On the server headerChan is always nil since a stream originates// only after having received headers.return}select {case <-s.ctx.Done():// Close the stream to prevent headers/trailers from changing after// this function returns.s.ct.CloseStream(s, ContextErr(s.ctx.Err()))// headerChan could possibly not be closed yet if closeStream raced// with operateHeaders; wait until it is closed explicitly here.<-s.headerChancase <-s.headerChan:}
}// 实为调用协程
0  0x000000000043bfa5 in runtime.goparkat /usr/local/go/src/runtime/proc.go:3071  0x000000000044c10f in runtime.selectgoat /usr/local/go/src/runtime/select.go:3382  0x000000000090c8d5 in google.golang/grpc/internal/transport.(*Stream).waitOnHeaderat /root/go/pkg/mod/google.golang/grpc@v1.33.2/internal/transport/transport.go:3213  0x0000000000942805 in google.golang/grpc/internal/transport.(*Stream).RecvCompressat /root/go/pkg/mod/google.golang/grpc@v1.33.2/internal/transport/transport.go:3364  0x0000000000942805 in google.golang/grpc.(*csAttempt).recvMsgat /root/go/pkg/mod/google.golang/grpc@v1.33.2/stream.go:8945  0x000000000094ad06 in google.golang/grpc.(*clientStream).RecvMsg.func1at /root/go/pkg/mod/google.golang/grpc@v1.33.2/stream.go:7596  0x000000000094057c in google.golang/grpc.(*clientStream).withRetryat /root/go/pkg/mod/google.golang/grpc@v1.33.2/stream.go:6177  0x0000000000941505 in google.golang/grpc.(*clientStream).RecvMsgat /root/go/pkg/mod/google.golang/grpc@v1.33.2/stream.go:7588  0x0000000000921d3b in google.golang/grpc.invokeat /root/go/pkg/mod/google.golang/grpc@v1.33.2/call.go:739  0x0000000000921ad3 in google.golang/grpc.(*ClientConn).Invokeat /root/go/pkg/mod/google.golang/grpc@v1.33.2/call.go:37
10  0x0000000000b185f4 in /root/hello/grpc/proto.(*HelloClient).Callat /root/hello/hello.pb.go:70

• 协程 poll_runtime_pollWait

// poll_runtime_pollWait, which is internal/poll.runtime_pollWait,
// waits for a descriptor to be ready for reading or writing,
// according to mode, which is 'r' or 'w'.
// This returns an error code; the codes are defined above.
//go:linkname poll_runtime_pollWait internal/poll.runtime_pollWait
func poll_runtime_pollWait(pd *pollDesc, mode int) int {errcode := netpollcheckerr(pd, int32(mode))if errcode != pollNoError {return errcode}// As for now only Solaris, illumos, and AIX use level-triggered IO.if GOOS == "solaris" || GOOS == "illumos" || GOOS == "aix" {netpollarm(pd, mode)}for !netpollblock(pd, int32(mode), false) {errcode = netpollcheckerr(pd, int32(mode))if errcode != pollNoError {return errcode}// Can happen if timeout has fired and unblocked us,// but before we had a chance to run, timeout has been reset.// Pretend it has not happened and retry.}return pollNoError
}函数 runtime.gopark 用于协程的切换
0  0x000000000043bfa5 in runtime.goparkat /usr/local/go/src/runtime/proc.go:307
1  0x000000000043447b in runtimepollblockat /usr/local/go/src/runtime/netpoll.go:4362  0x000000000046bdd5 in internal/poll.runtime_pollWaitat /usr/local/go/src/runtime/netpoll.go:220
3  0x00000000004d9685 in internal/poll.(*pollDesc).waitat /usr/local/go/src/internal/poll/fd_poll_runtime.go:87
4  0x00000000004da6c5 in internal/poll.(*pollDesc).waitReadat /usr/local/go/src/internal/poll/fd_poll_runtime.go:92
5  0x00000000004da6c5 in internal/poll.(*FD).Readat /usr/local/go/src/internal/poll/fd_unix.go:159
6  0x00000000005327af in net.(*netFD).Readat /usr/local/go/src/net/fd_posix.go:55
7  0x000000000054688e in net.(*conn).Readat /usr/local/go/src/net/net.go:182
8  0x00000000006e14b8 in net/http.(*connReader).backgroundReadat /usr/local/go/src/net/http/server.go:690
9  0x0000000000471821 in runtime.goexitat /usr/local/go/src/runtime/asm_amd64.s:1374

• 队列

type itemList struct {head *itemNodetail *itemNode
}type itemNode struct {it   interface{}next *itemNode
}// 入队（生产）
//
// 从这可看到，
// 队列没有大小限制，生产（入队）不受限，
// 所以一旦生产速度大于消费速度，就会出现堆积导致内存上涨。
func (il *itemList) enqueue(i interface{}) {n := &itemNode{it: i}if il.tail == nil {il.head, il.tail = n, nreturn}il.tail.next = nil.tail = n
}// 出队（消费）
func (il *itemList) dequeue() interface{} {if il.head == nil {return nil}i := il.head.itil.head = il.head.nextif il.head == nil {il.tail = nil}return i
}// 清空（消费），直接丢弃了
func (il *itemList) dequeueAll() *itemNode {h := il.headil.head, il.tail = nil, nilreturn h
}func (il *itemList) isEmpty() bool {return il.head == nil
}

// 源码所在文件：internal/transport/controlbuf.go
//    transport.loopyWriter.run
// -> transport.loopyWriter.handle
// -> transport.loopyWriter.headerHandler
// -> transport.loopyWriter.writeHeader // 阻塞在这了
func (l *loopyWriter) handle(i interface{}) error {switch i := i.(type) {case *incomingWindowUpdate:return l.incomingWindowUpdateHandler(i)case *outgoingWindowUpdate:return l.outgoingWindowUpdateHandler(i)case *incomingSettings:return l.incomingSettingsHandler(i)case *outgoingSettings:return l.outgoingSettingsHandler(i)case *headerFrame:return l.headerHandler(i) // 阻塞在这了......
}// 源码所在文件：internal/transport/controlbuf.go
func (l *loopyWriter) writeHeader(streamID uint32, endStream bool, hf []hpack.HeaderField, onWrite func()) error {// 阻塞在这儿：// 结构体 frame 定义在 internal/transport/http_util.go 文件中，// 成员 fr 的类型为 http2.Framer，定义在 x/net/http2/frame.go 文件中err = l.framer.fr.WriteHeaders(http2.HeadersFrameParam{})
}